Согласованность и репликация в Cassandra

← назад к разделу

Про Cassandra часто говорят «она жертвует согласованностью ради доступности». Это полуправда: согласованность в Cassandra настраивается, причём на каждый запрос отдельно. Устройство репликации по кольцу разбиралось в статье про архитектуру; здесь — про гарантии, которые из него складываются.

Replication factor: сколько копий

Replication factor (RF) — число копий каждой партиции в кластере. RF = 3 означает, что данные лежат на трёх разных узлах. Это фундамент: чем больше копий, тем выше надёжность и доступность, но тем дороже запись (её надо разнести по большему числу узлов).

RF задаётся на уровне пространства ключей (keyspace) и стратегии репликации. Для боевых систем берут NetworkTopologyStrategy, которая раскладывает копии с учётом дата-центров — например, по три копии в каждом из двух ЦОД.

Уровень согласованности: сколько реплик ответят

Лента активности переживёт, если реплика отдаст запись секундой позже; баланс счёта нет. Поэтому RF, число существующих копий, задают один раз, а сколько из них должны ответить, чтобы операция считалась успешной, выбирают на каждый запрос, отдельно для чтения и для записи: это уровень согласованности (consistency level, CL). Основные уровни:

ONE — быстро и доступно, но чтение с реплики, ещё не получившей обновление, вернёт старые данные. ALL — гарантирует свежесть, но стоит одной реплике упасть, и операция не проходит. Между ними — кворум.

Отдельно стоит запомнить, что получает тот, кто уровень вообще не выставлял. В драйвере DataStax 4.x по умолчанию стоит LOCAL_ONE — то есть самый слабый вариант, а не кворум. Код, который «просто пишет и читает», строгого чтения не получает, даже если вся остальная настройка кластера сделана правильно.

Главное правило: R + W > RF

Строгую (немедленную) согласованность в Cassandra дают не сами по себе уровни, а их сочетание при чтении и записи. Правило простое:

Если число реплик, подтверждающих запись (W), плюс число реплик, отвечающих на чтение (R), больше replication factor — чтение гарантированно увидит последнюю запись.

Почему: если R + W > RF, множества «кто подтвердил запись» и «кого спросили при чтении» обязательно пересекутся хотя бы на одной реплике — а значит, чтение зацепит свежую копию. Классический рецепт строгого чтения при RF = 3 — писать и читать с QUORUM (2 + 2 = 4 > 3). Если же и запись, и чтение идут с ONE (1 + 1 = 2 < 3), пересечение не гарантировано — получаем итоговую согласованность: данные сойдутся, но не мгновенно.

Главная ловушка: считать Cassandra «всегда несогласованной». При QUORUM/QUORUM она даёт строгое чтение — просто баланс скорости и свежести выбирают под каждый запрос: для баланса счёта кворум, для ленты активности хватит ONE.

Две оговорки, без которых правило подводит

Побеждает не «последняя» запись, а большая метка времени. Когда две реплики отдают разные значения, Cassandra не разбирается, какое из них новее по факту: у каждой записанной ячейки есть метка времени (timestamp), и выигрывает та, у которой метка больше. Это правило называют last-write-wins. Метку проставляет драйвер или узел-координатор — по своим часам. Значит, если часы на машинах разъехались хотя бы на секунду, «свежая» запись вполне может проиграть старой, и R + W > RF тут не спасёт: пересечение множеств выполнено, а выбрано всё равно не то. Если метки совпали в точности, база просто берёт большее значение побайтово — то есть исход зависит от содержимого, а не от порядка.

Провалившаяся запись не откатывается. Запись с QUORUM при RF = 3 вернёт ошибку, если подтвердили не две реплики, а одна. Но на этой одной реплике данные уже лежат — база их не убирает. Позже read repair или плановая сверка разнесут их по остальным копиям, и «неудавшаяся» запись всплывёт как удавшаяся. Поэтому фразу «чтение гарантированно увидит последнюю запись» читают буквально: про успешные записи она верна, а ошибка записи не означает «ничего не произошло».

В драйвере уровень ставится на сам запрос, а не на соединение целиком:

session.execute(insert.setConsistencyLevel(DefaultConsistencyLevel.QUORUM));
session.execute(select.setConsistencyLevel(DefaultConsistencyLevel.QUORUM));

if err := session.Query(insertCQL, args...).Consistency(gocql.Quorum).Exec(); err != nil {
	return err
}
iter := session.Query(selectCQL, chatID).Consistency(gocql.Quorum).Iter()

await client.execute(insert, params, { consistency: cassandra.types.consistencies.quorum });
const result = await client.execute(select, [chatId], { consistency: cassandra.types.consistencies.quorum });

insert.consistency_level = ConsistencyLevel.QUORUM
session.execute(insert, params)
select.consistency_level = ConsistencyLevel.QUORUM
rows = session.execute(select, (chat_id,))

Ту же арифметику можно проверить руками: запись обновляет первые W реплик из трёх, чтение опрашивает последние R (худший случай) и берёт самый свежий ответ:

import java.util.Arrays;

public class QuorumDemo {
    static final int RF = 3;

    static String readAfterWrite(int w, int r) {
        String[] replicas = new String[RF];
        Arrays.fill(replicas, "старое");
        for (int i = 0; i < w; i++) {
            replicas[i] = "свежее";
        }
        String seen = "старое";
        for (int i = RF - r; i < RF; i++) {
            if (replicas[i].equals("свежее")) {
                seen = "свежее";
            }
        }
        return seen;
    }

    public static void main(String[] args) {
        int[][] levels = {{2, 2}, {1, 1}, {1, 3}};
        for (int[] level : levels) {
            int w = level[0];
            int r = level[1];
            System.out.println("W=" + w + " R=" + r + ": R+W=" + (w + r)
                    + (w + r > RF ? " > " : " <= ") + RF
                    + " -> чтение видит " + readAfterWrite(w, r));
        }
    }
}

package main

import "fmt"

const rf = 3

func readAfterWrite(w, r int) string {
	replicas := make([]string, rf)
	for i := range replicas {
		replicas[i] = "старое"
	}
	for i := 0; i < w; i++ {
		replicas[i] = "свежее"
	}
	seen := "старое"
	for i := rf - r; i < rf; i++ {
		if replicas[i] == "свежее" {
			seen = "свежее"
		}
	}
	return seen
}

func main() {
	for _, level := range [][2]int{{2, 2}, {1, 1}, {1, 3}} {
		w, r := level[0], level[1]
		sign := " <= "
		if w+r > rf {
			sign = " > "
		}
		fmt.Printf("W=%d R=%d: R+W=%d%s%d -> чтение видит %s\n", w, r, w+r, sign, rf, readAfterWrite(w, r))
	}
}

const RF = 3;

function readAfterWrite(w, r) {
  const replicas = new Array(RF).fill('старое');
  for (let i = 0; i < w; i++) replicas[i] = 'свежее';
  let seen = 'старое';
  for (let i = RF - r; i < RF; i++) if (replicas[i] === 'свежее') seen = 'свежее';
  return seen;
}

for (const [w, r] of [[2, 2], [1, 1], [1, 3]]) {
  console.log(`W=${w} R=${r}: R+W=${w + r}${w + r > RF ? ' > ' : ' <= '}${RF} -> чтение видит ${readAfterWrite(w, r)}`);
}

RF = 3


def read_after_write(w: int, r: int) -> str:
    replicas = ["старое"] * RF
    for i in range(w):
        replicas[i] = "свежее"
    seen = "старое"
    for i in range(RF - r, RF):
        if replicas[i] == "свежее":
            seen = "свежее"
    return seen


for w, r in ((2, 2), (1, 1), (1, 3)):
    sign = " > " if w + r > RF else " <= "
    print(f"W={w} R={r}: R+W={w + r}{sign}{RF} -> чтение видит {read_after_write(w, r)}")

Сколько это стоит

У выбора уровня есть цена, и её полезно представлять хотя бы порядком величины.

ONE на чтение — один сетевой обмен с ближайшей репликой, задержка равна её времени ответа. QUORUM при трёх копиях ждёт две — то есть вторую по скорости реплику: задержка определяется не самой быстрой, а медленной из двух, и на практике это заметно дороже ONE, особенно на хвосте распределения (девяносто девятый процентиль). ALL ждёт самую медленную и падает, если недоступна хоть одна реплика, — его берут только осознанно и редко.

Облегчённая транзакция (IF NOT EXISTS, IF status = …) дороже в разы: вместо одной записи она проводит согласование по протоколу Paxos — обычно четыре сетевых обмена вместо одного. Отсюда правило: такими операциями закрывают уникальность и переходы статуса, а не обычную запись.

Между дата-центрами добавляется сетевая задержка: внутри одного ЦОД это доли миллисекунды, между регионами — десятки, а между континентами — больше сотни миллисекунд в одну сторону. Поэтому LOCAL_QUORUM отвечает за время, сравнимое с локальным запросом, а EACH_QUORUM — за время самого далёкого ЦОД плюс обмен.

Несколько дата-центров: какой уровень брать

Разница между уровнями здесь не в согласованности как таковой, а в том, чей отказ вас останавливает.

LOCAL_QUORUM требует большинство реплик в своём ЦОД. Пока живы два узла из трёх локально, приложение работает — даже если второй ЦОД полностью отключён. Это обычный выбор для мультирегиональной установки: каждый регион обслуживает своих пользователей и не зависит от связности между регионами.

EACH_QUORUM (только для записи) требует большинство в каждом ЦОД. Гарантия сильнее: после такой записи чтение с LOCAL_QUORUM в любом регионе увидит свежие данные. Цена — отказ одного ЦОД останавливает запись во всех, и задержка равна самому далёкому ЦОД.

Отсюда практическая раскладка. Пишем и читаем в своём регионе, между регионами допускаем отставание — LOCAL_QUORUM на обе операции (и R + W > RF считается внутри ЦОД). Нужна гарантия «записал в Москве, сразу прочитал во Франкфурте» — запись EACH_QUORUM, чтение LOCAL_QUORUM, и готовность к тому, что запись останавливается при потере связности. Пишем в оба ЦОД одновременно из разных приложений — тогда конфликты решаются по времени записи, и нужен либо один «пишущий» регион на каждый ключ, либо облегчённые транзакции.

Что прилетает, когда кворум не набрался

Ошибки стоит различать до того, как они появятся в проде, потому что реагировать на них надо по-разному.

UnavailableException — координатор заранее видит, что живых реплик меньше, чем требует уровень, и не начинает операцию. Данные точно не записаны, повтор безопасен (и уместен: возможно, узел вернулся).

WriteTimeoutException — операция началась, но нужное число подтверждений не пришло за таймаут. Ключевое: это не значит «не записано». Часть реплик могла записать, и данные потом разойдутся по кольцу сами. У исключения есть тип операции (writeType): SIMPLE и BATCH обычно можно повторять, потому что запись в Cassandra идемпотентна по ключу и времени; CAS (облегчённая транзакция) и COUNTER повторять вслепую нельзя — первая может уже применить изменение, второй прибавит второй раз.

ReadTimeoutException на чтении с флагом «данные получены, но не хватило подтверждений» означает, что ответ может быть неполным.

Отсюда правило обработки: повторять можно только идемпотентные операции, и повторять с разбросом паузы, а не сразу. Для неидемпотентных (счётчики, облегчённые транзакции) повтор заменяют проверкой состояния: прочитать и решить, применилось ли.

Спекулятивный повтор и таймауты

Первое, что видно в драйвере, когда одна реплика начала подтормаживать: запрос ждёт её, хотя рядом есть быстрые. Лечится это спекулятивным повтором — драйвер, не дождавшись ответа за заданное время, отправляет тот же запрос второй реплике и берёт ответ, который придёт раньше.

Настраивают это парой: политика спекулятивного повтора в драйвере (обычно по процентилю времени ответа) и таймауты — свой на запрос в драйвере и серверный (read_request_timeout, write_request_timeout). Правило согласования: таймаут драйвера должен быть больше серверного, иначе клиент отвалится раньше, чем сервер успеет ответить или вернуть свою ошибку, и вы получите загадочный обрыв вместо внятного исключения.

Оговорка: спекулятивный повтор тратит ресурсы кластера (тот же запрос выполняется дважды) и применим только к идемпотентным операциям — в драйвере запрос помечают явно (setIdempotent(true)).

Как реплики догоняют друг друга

Раз при ONE реплики временно расходятся, Cassandra их постоянно синхронизирует тремя механизмами:

• hinted handoff — если реплика в момент записи недоступна, координатор сохраняет «подсказку» (hint) и, когда реплика вернётся, доигрывает пропущенные записи. Так переживаются короткие сбои: подсказки копятся, пока узел лежит не дольше max_hint_window (по умолчанию три часа);
• read repair — при чтении с нескольких реплик координатор замечает расхождения и на лету обновляет отставшие копии самой свежей версией;
• anti-entropy repair — плановая фоновая сверка реплик (запускается администратором), которая находит и чинит расхождения, накопившиеся из-за долгих сбоев. Регулярный repair — обязательная часть эксплуатации Cassandra, и вот почему: удаление в Cassandra — это tombstone, пометка «этой строки больше нет», и живёт она ограниченный срок, gc_grace_seconds, по умолчанию десять дней. После этого пометку выбрасывают. Если реплика всё это время лежала и repair не прошёл, до неё пометка не доехала, а после срока её уже никто не принесёт — реплика вернётся в строй со «своей» копией удалённой строки, и удалённые данные воскреснут. Отсюда простое правило эксплуатации: repair прогоняют чаще, чем раз в gc_grace_seconds.

Вместе они и дают «итоговую согласованность»: разошедшиеся копии рано или поздно сходятся, а после долгого простоя узла их сводит только repair.

Чем именно запускается восстановление и что оно стоит — вопрос, который возникает первым в дежурство.

Команда одна: nodetool repair. Без параметров она сверяет все диапазоны токенов, которыми владеет узел, со всеми репликами — это самая дорогая операция в Cassandra: строятся деревья хешей по данным, сравниваются между репликами, расходящиеся куски передаются по сети. На больших таблицах восстановление идёт часами и заметно нагружает диск и сеть.

Отсюда варианты, которые применяют на практике. nodetool repair -pr (--partitioner-range) сверяет только тот диапазон, за который узел отвечает как основной, — тогда обход всего кольца делается запуском на каждом узле по очереди, без троекратного повторения одной работы. Инкрементальное восстановление помечает уже сверенные данные и в следующий раз проверяет только новые — дешевле, но у него своя история проблем с уплотнением, поэтому включают его осознанно. И полное восстановление раз в цикл gc_grace_seconds (по умолчанию десять суток) обязательно: если данные не сверить за этот срок, удалённые строки могут «воскреснуть», потому что метки удаления к тому времени уже убраны уплотнением.

В эксплуатации это оформляют расписанием: чаще всего инструментом-планировщиком (Reaper) или заданием, которое обходит узлы по одному в спокойные часы и следит, чтобы восстановление не шло одновременно с массовой загрузкой.

Облегчённые транзакции и их цена

А как же операции «сделать, только если ещё не сделано» — например, зарегистрировать пользователя с уникальным логином? Для таких случаев есть облегчённые транзакции (LWT, lightweight transactions) — условные операции вида INSERT ... IF NOT EXISTS или UPDATE ... IF column = ?. Под капотом они используют протокол консенсуса Paxos, чтобы обеспечить линеаризуемость в пределах одной партиции.

Но за это платят: LWT требует нескольких раундов между репликами и потому в разы медленнее обычной записи. Правило простое: LWT — для редких операций, где без строгого «сравни-и-запиши» никак (уникальность, защита от гонки), а не для массового потока. Если LWT нужны на каждом шагу — скорее всего, Cassandra выбрана не под ту задачу.

У LWT есть и своя пара уровней согласованности, отдельная от обычной. Она называется serial consistency и принимает два значения: SERIAL (кворум по всему кластеру) и LOCAL_SERIAL (кворум внутри своего ЦОД). Отсюда две вещи, которые новичка ловят. Первая: чтобы прочитать результат условной операции и не увидеть недорешённое состояние, читать надо тоже с SERIAL (или LOCAL_SERIAL) — обычный QUORUM этого не гарантирует. Вторая: смешивать на одной партиции условные и обычные записи не стоит — обычная запись идёт мимо Paxos и молча затрёт то, что условная операция только что аккуратно согласовала.

Чего Cassandra не гарантирует

Чтобы не ждать от базы того, чего в ней нет:

• нет транзакций между партициями. Атомарно изменить данные в разных партициях (тем более разных таблицах) нельзя. Пакет (BATCH) выглядит похоже на транзакцию, но ею не является. В пределах одной партиции он действительно применяется целиком. Между партициями обычный пакет (его называют LOGGED) гарантирует лишь, что рано или поздно применит все операции: для этого координатор сначала записывает пакет в служебный журнал и повторяет, пока не пройдёт. Но одновременности нет — читатель вполне может застать состояние, где половина изменений уже видна, а половина ещё нет. А есть второй вид, UNLOGGED BATCH, который берут ради скорости: у него служебного журнала нет вовсе, и обещания «рано или поздно применит всё» тоже нет — часть операций может просто не примениться, и никто об этом не узнает.
• нет отката и изоляции, как в SQL. Модель рассчитана на независимые записи по ключу, а не на сложные транзакционные сценарии.
• удаление — это запись (tombstone), поэтому «удалять и вставлять» массово — антипаттерн.

Всё это — не недостатки, а прямая плата за масштаб записи и постоянную доступность. Если эта плата неприемлема — значит, задаче нужна PostgreSQL или MongoDB, а не Cassandra.

Глубже: Cassandra из приложения: драйвер, уровень согласованности и подготовленные запросырасширенное

spring:
  cassandra:
    contact-points: cass-1,cass-2
    local-datacenter: dc1
    keyspace-name: chat
    request:
      consistency: local-quorum
      timeout: 2s

PreparedStatement byChat = session.prepare(
    "SELECT sent_at, author, body FROM messages WHERE chat_id = ? AND sent_at > ?");

ResultSet rows = session.execute(byChat.bind(chatId, since));

cluster := gocql.NewCluster("cass-1", "cass-2")
cluster.Keyspace = "chat"
cluster.Consistency = gocql.LocalQuorum
cluster.Timeout = 2 * time.Second
cluster.PoolConfig.HostSelectionPolicy = gocql.TokenAwareHostPolicy(gocql.DCAwareRoundRobinPolicy("dc1"))
session, err := cluster.CreateSession()

iter := session.Query(
	"SELECT sent_at, author, body FROM messages WHERE chat_id = ? AND sent_at > ?",
	chatID, since).Iter()
var sentAt time.Time
var author, body string
for iter.Scan(&sentAt, &author, &body) {
	handle(sentAt, author, body)
}
if err := iter.Close(); err != nil {
	return err
}

const cassandra = require('cassandra-driver');

const client = new cassandra.Client({
  contactPoints: ['cass-1', 'cass-2'],
  localDataCenter: 'dc1',
  keyspace: 'chat',
  queryOptions: { consistency: cassandra.types.consistencies.localQuorum, prepare: true },
  socketOptions: { readTimeout: 2000 },
});
await client.connect();

const byChat = 'SELECT sent_at, author, body FROM messages WHERE chat_id = ? AND sent_at > ?';

const result = await client.execute(byChat, [chatId, since], { prepare: true });
for (const row of result.rows) {
  handle(row.sent_at, row.author, row.body);
}

from cassandra import ConsistencyLevel
from cassandra.cluster import EXEC_PROFILE_DEFAULT, Cluster, ExecutionProfile
from cassandra.policies import DCAwareRoundRobinPolicy, TokenAwarePolicy

profile = ExecutionProfile(
    load_balancing_policy=TokenAwarePolicy(DCAwareRoundRobinPolicy(local_dc="dc1")),
    consistency_level=ConsistencyLevel.LOCAL_QUORUM,
    request_timeout=2,
)
cluster = Cluster(["cass-1", "cass-2"], execution_profiles={EXEC_PROFILE_DEFAULT: profile})
session = cluster.connect("chat")

by_chat = session.prepare(
    "SELECT sent_at, author, body FROM messages WHERE chat_id = ? AND sent_at > ?")

for row in session.execute(by_chat, (chat_id, since)):
    handle(row.sent_at, row.author, row.body)

Коротко

• Replication factor (RF) — сколько копий партиции существует; уровень согласованности (CL) — сколько реплик должны ответить, задаётся на каждый запрос отдельно для чтения и записи.
• Строгое чтение даёт правило R + W > RF (классика — QUORUM на запись и чтение при RF = 3); ONE/ONE даёт итоговую согласованность. Cassandra не «несогласованная», а настраиваемая. По умолчанию драйвер DataStax 4.x ставит LOCAL_ONE — уровень надо выставлять осознанно.
• У правила две оговорки: конфликт решается по метке времени (last-write-wins), поэтому разъехавшиеся часы ломают «свежесть»; и провалившаяся запись не откатывается — она могла лечь на одну реплику и позже разойтись по остальным.
• Реплики сводятся через hinted handoff, read repair и плановый anti-entropy repair: последний запускают nodetool repair (обычно -pr по узлам, по расписанию) обязательно чаще, чем gc_grace_seconds — иначе удалённые данные воскресают.
• LWT (Paxos) дают строгое «сравни-и-запиши» в пределах партиции ценой примерно четырёх сетевых обменов вместо одного — только для редких операций; читать их результат надо с SERIAL / LOCAL_SERIAL, а обычные записи на ту же партицию не подмешивать.
• Нет транзакций между партициями, нет отката/изоляции как в SQL, удаление — это tombstone. BATCH транзакцию не заменяет, а UNLOGGED BATCH не даёт даже обещания «применится всё». Это плата за масштаб.
• Из приложения: сессия драйвера с явным локальным дата-центром, уровень согласованности в настройках драйвера (по умолчанию слабый LOCAL_ONE, у gocql QUORUM), запросы только подготовленные, повторы драйвера лишь для идемпотентных запросов, тесты через Testcontainers.
• QUORUM ждёт вторую по скорости реплику, а между регионами добавляются десятки миллисекунд в одну сторону — отсюда и спекулятивный повтор в драйвере для идемпотентных запросов.
• LOCAL_QUORUM переживает отказ соседнего ЦОД, EACH_QUORUM даёт «записал здесь — прочитал там» ценой остановки записи при потере связности.
• UnavailableException означает «не начиналось», WriteTimeoutException — «может быть записано»; повторять можно только идемпотентное, а счётчики и облегчённые транзакции — нет.

Что почитать дальше

• Архитектура Cassandra: кольцо без главного узла — как устроены кольцо, репликация и путь записи, на которых стоят эти гарантии.
• Модель данных Cassandra: partition key и clustering columns — как проектировать таблицы под доступ по ключу раздела.
• Cassandra, PostgreSQL или MongoDB: когда брать колоночную NoSQL — когда настраиваемая согласованность и масштаб записи оправданы, а когда нужна классическая база.
• Модели репликации простыми словами - откуда берутся кворумы и чем опасно отставание реплики, разобрано без привязки к Cassandra.