Apache Cassandra: первый взгляд системного архитектора

CREATE TABLE users (
    user_id UUID,
    email TEXT,
    created_at TIMESTAMP,
    PRIMARY KEY (user_id)  -- user_id — partition key
);
 
-- Все записи для user_id=123 будут на одних и тех же узлах
-- Cassandra хеширует user_id и решает, какие узлы хранят эти данные

CREATE TABLE user_events (
    user_id UUID,
    event_time TIMESTAMP,
    event_type TEXT,
    payload TEXT,
    PRIMARY KEY (user_id, event_time)
    -- user_id — partition key (на каких узлах хранить)
    -- event_time — clustering key (как сортировать внутри партиции)
) WITH CLUSTERING ORDER BY (event_time DESC);
 
-- Все события user_id=123 хранятся вместе (одна партиция)
-- Внутри партиции отсортированы по event_time от новых к старым
-- Запрос последних 100 событий — O(1) чтение, без сканирования всей таблицы

# Запись с CL=QUORUM: ждём подтверждения от большинства реплик
session.execute(
    insert_query,
    consistency_level=ConsistencyLevel.QUORUM  # 2 из 3 для RF=3
)
 
# Запись с CL=ONE: ждём подтверждения от любой одной реплики
session.execute(
    insert_query,
    consistency_level=ConsistencyLevel.ONE  # Быстро, но риск неконсистентности
)
 
# Чтение с CL=ALL: ждём ответа от всех реплик
session.execute(
    select_query,
    consistency_level=ConsistencyLevel.ALL  # Медленно, но самые свежие данные
)

-- PostgreSQL: плохо масштабируется для time-series
CREATE TABLE device_metrics (
    device_id UUID,
    metric_time TIMESTAMP,
    cpu_usage FLOAT,
    memory_usage FLOAT,
    PRIMARY KEY (device_id, metric_time)
);
 
-- Проблемы:
-- 1. INSERT нагрузка упирается в один master
-- 2. Партиционирование по времени нужно создавать вручную
-- 3. Запросы к старым партициям медленные
-- 4. Шардинг по device_id требует прокси-слой (Citus, Vitess)

-- Cassandra: идеально для time-series
CREATE TABLE device_metrics (
    device_id UUID,
    metric_time TIMESTAMP,
    cpu_usage FLOAT,
    memory_usage FLOAT,
    PRIMARY KEY (device_id, metric_time)
) WITH CLUSTERING ORDER BY (metric_time DESC);
 
-- Преимущества:
-- 1. INSERT распределяются по всем узлам автоматически
-- 2. Данные каждого device_id хранятся вместе (одна партиция)
-- 3. Внутри партиции метрики отсортированы по времени
-- 4. Запрос последних 1000 метрик устройства — O(1) чтение
-- 5. Добавили узел → write throughput вырос пропорционально

from cassandra.cluster import Cluster
from datetime import datetime
import uuid
 
# Подключение к кластеру
cluster = Cluster(['192.168.1.10', '192.168.1.11', '192.168.1.12'])
session = cluster.connect('monitoring')
 
# Запись метрик — распределяется по узлам автоматически
device_id = uuid.uuid4()
for i in range(1000):
    session.execute(
        """
        INSERT INTO device_metrics (device_id, metric_time, cpu_usage, memory_usage)
        VALUES (%s, %s, %s, %s)
        """,
        (device_id, datetime.utcnow(), 45.2 + i * 0.1, 62.8 + i * 0.05)
    )
 
# Чтение последних 100 метрик — O(1) благодаря clustering key
rows = session.execute(
    """
    SELECT metric_time, cpu_usage, memory_usage
    FROM device_metrics
    WHERE device_id = %s
    LIMIT 100
    """,
    (device_id,)
)
 
# Все 100 записей находятся на одних узлах, отсортированы — быстрое чтение
for row in rows:
    print(f"{row.metric_time}: CPU {row.cpu_usage}%, MEM {row.memory_usage}%")

# Cassandra для логирования
CREATE TABLE application_logs (
    app_name TEXT,
    log_date DATE,
    log_time TIMESTAMP,
    level TEXT,
    message TEXT,
    PRIMARY KEY ((app_name, log_date), log_time)
) WITH CLUSTERING ORDER BY (log_time DESC);
 
# Composite partition key (app_name, log_date):
# - Логи каждого приложения за день хранятся вместе
# - Автоматическое "партиционирование по дням" без ручного создания партиций
# - Запросы по приложению и дате — O(1) чтение с одной партиции

-- Cassandra имеет нативные counter-колонки
CREATE TABLE article_stats (
    article_id UUID PRIMARY KEY,
    views COUNTER,
    clicks COUNTER,
    shares COUNTER
);
 
-- Инкремент без блокировок
UPDATE article_stats
SET views = views + 1
WHERE article_id = 550e8400-e29b-41d4-a716-446655440000;

-- Cassandra для сессий
CREATE TABLE user_sessions (
    session_id UUID PRIMARY KEY,
    user_id UUID,
    data TEXT,  -- JSON с данными сессии
    expires_at TIMESTAMP
) WITH default_time_to_live = 86400;  -- TTL 24 часа
 
-- Преимущества:
-- 1. Персистентность на диск (не потеряем сессии при рестарте)
-- 2. Автоматическое удаление через TTL
-- 3. Линейное масштабирование — добавили узел, больше памяти для сессий
-- 4. Географическое распределение — сессии реплицируются между дата-центрами

-- Cassandra для простых очередей
CREATE TABLE task_queue (
    queue_name TEXT,
    task_id TIMEUUID,
    payload TEXT,
    status TEXT,
    PRIMARY KEY (queue_name, task_id)
) WITH CLUSTERING ORDER BY (task_id ASC);
 
-- Enqueue
INSERT INTO task_queue (queue_name, task_id, payload, status)
VALUES ('email_notifications', now(), '{"to": "user@example.com"}', 'pending');
 
-- Dequeue (atomic update)
UPDATE task_queue
SET status = 'processing'
WHERE queue_name = 'email_notifications'
  AND task_id = <минимальный pending task_id>
IF status = 'pending';  -- LWT для атомарности

services:
  cassandra:
    image: cassandra:5.0
    container_name: cassandra-node1
    ports:
      - "9042:9042" # CQL порт
      - "7199:7199" # JMX мониторинг
    environment:
      - CASSANDRA_CLUSTER_NAME=DevCluster
      - CASSANDRA_DC=datacenter1
      - CASSANDRA_RACK=rack1
      - CASSANDRA_ENDPOINT_SNITCH=GossipingPropertyFileSnitch
      - MAX_HEAP_SIZE=512M
      - HEAP_NEWSIZE=128M
    volumes:
      - cassandra_data:/var/lib/cassandra
    healthcheck:
      test: ["CMD-SHELL", "cqlsh -e 'describe cluster'"]
      interval: 30s
      timeout: 10s
      retries: 5
 
volumes:
  cassandra_data:

# Поднимаем Cassandra
docker compose up -d
 
# Проверяем статус
docker compose ps
 
# Логи
docker compose logs -f cassandra
 
# Подключаемся через cqlsh (CQL shell)
docker exec -it cassandra-node1 cqlsh
 
# В cqlsh проверяем кластер
cqlsh> DESCRIBE CLUSTER;
cqlsh> SELECT * FROM system.local;

-- Создаём keyspace с SimpleStrategy (для dev)
CREATE KEYSPACE IF NOT EXISTS demo
WITH replication = {
    'class': 'SimpleStrategy',
    'replication_factor': 1
};
 
-- В production используйте NetworkTopologyStrategy
CREATE KEYSPACE IF NOT EXISTS production_app
WITH replication = {
    'class': 'NetworkTopologyStrategy',
    'datacenter1': 3  -- 3 копии данных в datacenter1
};
 
-- Используем keyspace
USE demo;
 
-- Создаём таблицу
CREATE TABLE users (
    user_id UUID PRIMARY KEY,
    email TEXT,
    full_name TEXT,
    created_at TIMESTAMP,
    last_login TIMESTAMP
);
 
-- Создаём таблицу с composite primary key
CREATE TABLE user_sessions (
    user_id UUID,
    session_id TIMEUUID,
    ip_address TEXT,
    user_agent TEXT,
    created_at TIMESTAMP,
    PRIMARY KEY (user_id, session_id)
) WITH CLUSTERING ORDER BY (session_id DESC);

-- INSERT
INSERT INTO users (user_id, email, full_name, created_at)
VALUES (uuid(), 'constantin@potapov.me', 'Constantin Potapov', toTimestamp(now()));
 
-- SELECT
SELECT * FROM users WHERE user_id = 550e8400-e29b-41d4-a716-446655440000;
 
-- UPDATE
UPDATE users
SET last_login = toTimestamp(now())
WHERE user_id = 550e8400-e29b-41d4-a716-446655440000;
 
-- DELETE
DELETE FROM users WHERE user_id = 550e8400-e29b-41d4-a716-446655440000;
 
-- Batch операции (в пределах одной партиции!)
BEGIN BATCH
    INSERT INTO users (user_id, email, full_name) VALUES (uuid(), 'user1@ex.com', 'User 1');
    INSERT INTO users (user_id, email, full_name) VALUES (uuid(), 'user2@ex.com', 'User 2');
APPLY BATCH;

pip install cassandra-driver

from cassandra.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT
from cassandra.policies import DCAwareRoundRobinPolicy, TokenAwarePolicy
from cassandra.query import dict_factory
import uuid
 
# Настройка профиля выполнения
profile = ExecutionProfile(
    load_balancing_policy=TokenAwarePolicy(
        DCAwareRoundRobinPolicy(local_dc='datacenter1')
    ),
    row_factory=dict_factory  # Возвращать строки как dict
)
 
# Подключение к кластеру
cluster = Cluster(
    contact_points=['192.168.1.10', '192.168.1.11', '192.168.1.12'],
    execution_profiles={EXEC_PROFILE_DEFAULT: profile},
    protocol_version=4,
)
 
session = cluster.connect('demo')
 
# Prepared statement для производительности
insert_user = session.prepare(
    """
    INSERT INTO users (user_id, email, full_name, created_at)
    VALUES (?, ?, ?, toTimestamp(now()))
    """
)
 
# Вставка данных
user_id = uuid.uuid4()
session.execute(insert_user, (user_id, 'test@example.com', 'Test User'))
 
# Чтение данных
rows = session.execute(
    "SELECT * FROM users WHERE user_id = %s",
    (user_id,)
)
 
for row in rows:
    print(f"User: {row['full_name']}, Email: {row['email']}")
 
# Закрываем соединение
cluster.shutdown()

pip install aiocassandra

from fastapi import FastAPI, HTTPException
from cassandra.cluster import Cluster
from cassandra.query import dict_factory
from aiocassandra import aiosession
from pydantic import BaseModel
from typing import Optional
import uuid
 
app = FastAPI()
 
# Инициализация Cassandra при старте
cluster = Cluster(['192.168.1.10'])
session = cluster.connect('demo')
session.row_factory = dict_factory
 
# Делаем session асинхронным
async_session = aiosession(session)
 
class User(BaseModel):
    email: str
    full_name: str
 
class UserResponse(User):
    user_id: uuid.UUID
    created_at: Optional[str]
 
@app.on_event("shutdown")
def shutdown_event():
    cluster.shutdown()
 
@app.post("/users", response_model=UserResponse)
async def create_user(user: User):
    """Создание пользователя"""
    user_id = uuid.uuid4()
 
    query = """
    INSERT INTO users (user_id, email, full_name, created_at)
    VALUES (?, ?, ?, toTimestamp(now()))
    """
 
    await async_session.execute(query, (user_id, user.email, user.full_name))
 
    return UserResponse(
        user_id=user_id,
        email=user.email,
        full_name=user.full_name,
        created_at=None
    )
 
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: uuid.UUID):
    """Получение пользователя"""
    rows = await async_session.execute(
        "SELECT * FROM users WHERE user_id = %s",
        (user_id,)
    )
 
    users = list(rows)
    if not users:
        raise HTTPException(status_code=404, detail="User not found")
 
    return UserResponse(**users[0])

# ❌ Плохо: парсинг CQL на каждый запрос
session.execute(
    f"INSERT INTO users (user_id, email) VALUES ({user_id}, '{email}')"
)
 
# ✅ Хорошо: подготовили один раз, используем многократно
insert_stmt = session.prepare(
    "INSERT INTO users (user_id, email, full_name) VALUES (?, ?, ?)"
)
 
# Prepared statement парсится один раз, затем Cassandra знает структуру запроса
# Вместо парсинга SQL на каждую вставку — подстановка параметров
for user in users:
    session.execute(insert_stmt, (user.id, user.email, user.name))

-- ❌ Антипаттерн: пытаемся делать JOIN через WHERE
CREATE TABLE articles (
    article_id UUID PRIMARY KEY,
    author_id UUID,
    title TEXT,
    content TEXT,
    published_at TIMESTAMP
);
 
CREATE TABLE comments (
    comment_id UUID PRIMARY KEY,
    article_id UUID,
    author_id UUID,
    text TEXT
);
 
-- Проблема 1: Нельзя эффективно получить статьи автора
-- SELECT * FROM articles WHERE author_id = ? ALLOW FILTERING;
-- ALLOW FILTERING сканирует ВСЕ партиции — катастрофа для производительности!
 
-- Проблема 2: Нет сортировки по published_at без ALLOW FILTERING

-- Запрос 1: статья по ID
CREATE TABLE articles_by_id (
    article_id UUID PRIMARY KEY,
    author_id UUID,
    title TEXT,
    content TEXT,
    published_at TIMESTAMP
);
 
-- Запрос 2: статьи автора, отсортированные по дате
CREATE TABLE articles_by_author (
    author_id UUID,
    published_at TIMESTAMP,
    article_id UUID,
    title TEXT,
    content TEXT,  -- Денормализация: дублируем content
    PRIMARY KEY (author_id, published_at)
) WITH CLUSTERING ORDER BY (published_at DESC);
 
-- Запрос 3: последние статьи (главная страница)
CREATE TABLE articles_recent (
    bucket TEXT,  -- Например 'home_page'
    published_at TIMESTAMP,
    article_id UUID,
    author_id UUID,
    title TEXT,
    PRIMARY KEY (bucket, published_at)
) WITH CLUSTERING ORDER BY (published_at DESC);
 
-- Запрос 4: комментарии к статье
CREATE TABLE comments_by_article (
    article_id UUID,
    comment_id TIMEUUID,  -- TIMEUUID для сортировки по времени создания
    author_id UUID,
    text TEXT,
    PRIMARY KEY (article_id, comment_id)
) WITH CLUSTERING ORDER BY (comment_id DESC);

from cassandra.cluster import Cluster
import uuid
 
cluster = Cluster(['192.168.1.10'])
session = cluster.connect('blog')
 
# Публикация статьи — пишем в ТРИ таблицы (денормализация)
article_id = uuid.uuid4()
author_id = uuid.uuid4()
title = "Apache Cassandra: первый взгляд"
content = "Экспертный разбор Cassandra..."
published_at = datetime.utcnow()
 
# Таблица 1: для запроса по ID
session.execute(
    "INSERT INTO articles_by_id (article_id, author_id, title, content, published_at) VALUES (?, ?, ?, ?, ?)",
    (article_id, author_id, title, content, published_at)
)
 
# Таблица 2: для запроса статей автора
session.execute(
    "INSERT INTO articles_by_author (author_id, published_at, article_id, title, content) VALUES (?, ?, ?, ?, ?)",
    (author_id, published_at, article_id, title, content)
)
 
# Таблица 3: для главной страницы
session.execute(
    "INSERT INTO articles_recent (bucket, published_at, article_id, author_id, title) VALUES (?, ?, ?, ?, ?)",
    ('home_page', published_at, article_id, author_id, title)
)
 
# Чтение статей автора — O(1) по partition key
rows = session.execute(
    "SELECT * FROM articles_by_author WHERE author_id = ? LIMIT 20",
    (author_id,)
)
 
# Чтение комментариев к статье — O(1) по partition key
rows = session.execute(
    "SELECT * FROM comments_by_article WHERE article_id = ?",
    (article_id,)
)

-- ❌ Плохо: все события всех пользователей в одной партиции
CREATE TABLE all_events (
    event_type TEXT,  -- partition key
    event_id TIMEUUID,
    user_id UUID,
    payload TEXT,
    PRIMARY KEY (event_type, event_id)
);
-- Проблема: event_type='login' может содержать миллиарды записей → гигантская партиция
 
-- ✅ Хорошо: партиция на каждого пользователя
CREATE TABLE user_events (
    user_id UUID,     -- partition key
    event_id TIMEUUID,
    event_type TEXT,
    payload TEXT,
    PRIMARY KEY (user_id, event_id)
);
-- user_id разбивает данные на миллионы маленьких партиций

CREATE TABLE messages (
    chat_id UUID,
    message_id TIMEUUID,  -- Содержит timestamp создания
    sender_id UUID,
    text TEXT,
    PRIMARY KEY (chat_id, message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);
 
-- Преимущества:
-- 1. Автоматическая сортировка по времени создания
-- 2. Уникальность даже при миллионах записей в секунду
-- 3. Можно извлечь timestamp: SELECT dateOf(message_id) FROM messages

-- Вставка с TTL (Time To Live) 24 часа
INSERT INTO user_sessions (session_id, user_id, data)
VALUES (uuid(), uuid(), '{"key": "value"}')
USING TTL 86400;
 
-- Cassandra автоматически удалит эту запись через 24 часа
-- Нет нужды в cron-джобах для очистки старых данных

-- DELETE создаёт tombstone
DELETE FROM users WHERE user_id = ?;
 
-- Tombstone остаётся на диске до compaction (по умолчанию 10 дней)
-- Если удалили миллион записей → миллион tombstones
-- При SELECT Cassandra читает tombstones → замедление запросов

-- Вместо DELETE через 24 часа используем TTL
INSERT INTO sessions (session_id, data) VALUES (?, ?)
USING TTL 86400;
 
-- Cassandra сама удалит через 24 часа, tombstones минимальны

-- По умолчанию tombstones живут 10 дней (864000 секунд)
-- Если не используете ручные repair, можно сократить до 1 дня
ALTER TABLE sessions WITH gc_grace_seconds = 86400;

# cassandra.yaml
tombstone_warn_threshold: 1000 # Предупреждение при 1000 tombstones в запросе
tombstone_failure_threshold: 100000 # Ошибка при 100k tombstones

-- Создали таблицу с partition key = user_id
CREATE TABLE users (
    user_id UUID PRIMARY KEY,
    email TEXT,
    full_name TEXT
);
 
-- Хотим искать по email → создаём secondary index
CREATE INDEX ON users (email);
 
-- Запрос работает
SELECT * FROM users WHERE email = 'test@example.com';

-- Таблица для запросов по email
CREATE TABLE users_by_email (
    email TEXT PRIMARY KEY,
    user_id UUID,
    full_name TEXT
);
 
-- Теперь запрос по email — O(1), использует partition key
SELECT * FROM users_by_email WHERE email = 'test@example.com';

-- ❌ Ошибка
SELECT * FROM users WHERE full_name = 'John';
-- Error: Cannot execute this query as it might involve data filtering
 
-- Cassandra предлагает добавить ALLOW FILTERING:
SELECT * FROM users WHERE full_name = 'John' ALLOW FILTERING;

CREATE TABLE users_by_name (
    full_name TEXT,
    user_id UUID,
    email TEXT,
    PRIMARY KEY (full_name, user_id)
);

-- ❌ Плохой partition key: большинство запросов к одному значению
CREATE TABLE page_views (
    page_url TEXT,  -- partition key
    view_id TIMEUUID,
    user_id UUID,
    PRIMARY KEY (page_url, view_id)
);
 
-- Проблема: главная страница '/' получает 90% трафика
-- Партиция '/' перегружена, остальные узлы простаивают

-- ✅ Хороший подход: composite partition key с bucketing
CREATE TABLE page_views (
    page_url TEXT,
    bucket INT,  -- Случайное число 0-99 (синтетический ключ для распределения)
    view_id TIMEUUID,
    user_id UUID,
    PRIMARY KEY ((page_url, bucket), view_id)
);
 
-- Теперь главная страница распределена по 100 партиций:
-- ('/', 0), ('/', 1), ..., ('/', 99)
-- Нагрузка равномерна по узлам — каждая партиция получает ~1% трафика

import random
 
page_url = '/'
bucket = random.randint(0, 99)  # Случайная корзина 0-99
 
session.execute(
    "INSERT INTO page_views (page_url, bucket, view_id, user_id) VALUES (?, ?, ?, ?)",
    (page_url, bucket, uuid.uuid1(), user_id)
)

# Читаем из всех 100 корзин параллельно
from concurrent.futures import ThreadPoolExecutor
 
def read_bucket(bucket):
    return session.execute(
        "SELECT * FROM page_views WHERE page_url = ? AND bucket = ? LIMIT 100",
        (page_url, bucket)
    )
 
with ThreadPoolExecutor(max_workers=10) as executor:
    results = executor.map(read_bucket, range(100))
    all_views = [view for bucket_views in results for view in bucket_views]

# ❌ Антипаттерн: batch для разных партиций
batch = BatchStatement()
for user in users:
    batch.add(insert_stmt, (user.id, user.email, user.name))
session.execute(batch)
 
# Проблема: Cassandra должна координировать запись во все партиции
# Coordinator узел становится узким местом

# ✅ Правильно: параллельные асинхронные вставки
from cassandra.query import SimpleStatement
 
futures = []
for user in users:
    future = session.execute_async(insert_stmt, (user.id, user.email, user.name))
    futures.append(future)
 
# Ждём завершения всех операций
for future in futures:
    future.result()

BEGIN BATCH
    INSERT INTO users_by_id (user_id, email) VALUES (?, ?);
    INSERT INTO users_by_email (email, user_id) VALUES (?, ?);
APPLY BATCH;
 
-- Обе записи имеют одинаковый partition key (user_id или email)
-- Cassandra гарантирует атомарность

# docker-compose.yml
services:
  cassandra:
    image: cassandra:5.0
    environment:
      - JVM_OPTS=-javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent.jar=7070:/opt/jmx_exporter/config.yml
    ports:
      - "7070:7070" # Prometheus metrics
    volumes:
      - ./jmx_exporter:/opt/jmx_exporter
 
  prometheus:
    image: prom/prometheus
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - "9090:9090"
 
  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin

scrape_configs:
  - job_name: "cassandra"
    static_configs:
      - targets: ["cassandra:7070"]

# cassandra.yaml
slow_query_log_timeout: 500ms # Логировать запросы дольше 500ms

# cassandra.yaml
authenticator: PasswordAuthenticator # Вместо AllowAllAuthenticator
authorizer: CassandraAuthorizer # Вместо AllowAllAuthorizer
 
# Включите SSL для client-to-node
client_encryption_options:
  enabled: true
  keystore: /path/to/keystore.jks
  keystore_password: changeit
 
# Включите SSL для node-to-node
server_encryption_options:
  internode_encryption: all
  keystore: /path/to/keystore.jks
  keystore_password: changeit

-- Создаём роль для приложения
CREATE ROLE app_user WITH PASSWORD = 'secure_password' AND LOGIN = true;
 
-- Даём права только на конкретный keyspace
GRANT SELECT, MODIFY ON KEYSPACE production_app TO app_user;
 
-- Запрещаем DROP и TRUNCATE
REVOKE DROP ON KEYSPACE production_app FROM app_user;