Apache Cassandra: A System Architect's First Look

CREATE TABLE users (
    user_id UUID,
    email TEXT,
    created_at TIMESTAMP,
    PRIMARY KEY (user_id)  -- user_id is the partition key
);
 
-- All rows for user_id=123 will be on the same nodes
-- Cassandra hashes user_id and decides which nodes store these data

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 (which nodes store the data)
    -- event_time = clustering key (how to sort inside the partition)
) WITH CLUSTERING ORDER BY (event_time DESC);
 
-- All events for user_id=123 live together (one partition)
-- Inside the partition they are sorted by event_time from new to old
-- Fetching the last 100 events is O(1) read without scanning the whole table

# Write with CL=QUORUM: wait for confirmation from the majority of replicas
session.execute(
    insert_query,
    consistency_level=ConsistencyLevel.QUORUM  # 2 of 3 for RF=3
)
 
# Write with CL=ONE: wait for any one replica
session.execute(
    insert_query,
    consistency_level=ConsistencyLevel.ONE  # Fast but risk of inconsistency
)
 
# Read with CL=ALL: wait for all replicas
session.execute(
    select_query,
    consistency_level=ConsistencyLevel.ALL  # Slow but freshest data
)

-- PostgreSQL: scales poorly for time-series
CREATE TABLE device_metrics (
    device_id UUID,
    metric_time TIMESTAMP,
    cpu_usage FLOAT,
    memory_usage FLOAT,
    PRIMARY KEY (device_id, metric_time)
);
 
-- Problems:
-- 1. INSERT load hits a single master
-- 2. Time-based partitioning must be created manually
-- 3. Queries on old partitions are slow
-- 4. Sharding by device_id needs a proxy layer (Citus, Vitess)

-- Cassandra: ideal for 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);
 
-- Advantages:
-- 1. INSERTs are distributed across all nodes automatically
-- 2. Data for each device_id live together (one partition)
-- 3. Inside the partition metrics are sorted by time
-- 4. Fetching the latest 1000 metrics of a device - O(1) read
-- 5. Add a node -> write throughput grows proportionally

from cassandra.cluster import Cluster
from datetime import datetime
import uuid
 
# Connect to the cluster
cluster = Cluster(['192.168.1.10', '192.168.1.11', '192.168.1.12'])
session = cluster.connect('monitoring')
 
# Write metrics - automatically distributed across nodes
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)
    )
 
# Read the last 100 metrics - O(1) thanks to the clustering key
rows = session.execute(
    """
    SELECT metric_time, cpu_usage, memory_usage
    FROM device_metrics
    WHERE device_id = %s
    LIMIT 100
    """,
    (device_id,)
)
 
# All 100 rows are on the same nodes and sorted - fast read
for row in rows:
    print(f"{row.metric_time}: CPU {row.cpu_usage}%, MEM {row.memory_usage}%")

# Cassandra for logging
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):
# - Logs of each app for a day stay together
# - Automatic "partition by day" without manual creation
# - Queries by app and date - O(1) read from a single partition

-- Cassandra has native counter columns
CREATE TABLE article_stats (
    article_id UUID PRIMARY KEY,
    views COUNTER,
    clicks COUNTER,
    shares COUNTER
);
 
-- Increment without locks
UPDATE article_stats
SET views = views + 1
WHERE article_id = 550e8400-e29b-41d4-a716-446655440000;

-- Cassandra for sessions
CREATE TABLE user_sessions (
    session_id UUID PRIMARY KEY,
    user_id UUID,
    data TEXT,  -- JSON with session data
    expires_at TIMESTAMP
) WITH default_time_to_live = 86400;  -- TTL 24 hours
 
-- Advantages:
-- 1. Persistence on disk (sessions survive restarts)
-- 2. Automatic deletion via TTL
-- 3. Linear scaling - add a node, more memory for sessions
-- 4. Geographic distribution - sessions replicate between data centers

-- Cassandra for simple queues
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 = <smallest pending task_id>
IF status = 'pending';  -- LWT for atomicity

services:
  cassandra:
    image: cassandra:5.0
    container_name: cassandra-node1
    ports:
      - "9042:9042" # CQL port
      - "7199:7199" # JMX monitoring
    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:

# Start Cassandra
docker compose up -d
 
# Check status
docker compose ps
 
# Logs
docker compose logs -f cassandra
 
# Connect via cqlsh (CQL shell)
docker exec -it cassandra-node1 cqlsh
 
# Check the cluster in cqlsh
cqlsh> DESCRIBE CLUSTER;
cqlsh> SELECT * FROM system.local;

-- Create a keyspace with SimpleStrategy (for dev)
CREATE KEYSPACE IF NOT EXISTS demo
WITH replication = {
    'class': 'SimpleStrategy',
    'replication_factor': 1
};
 
-- In production use NetworkTopologyStrategy
CREATE KEYSPACE IF NOT EXISTS production_app
WITH replication = {
    'class': 'NetworkTopologyStrategy',
    'datacenter1': 3  -- 3 copies of data in datacenter1
};
 
-- Use the keyspace
USE demo;
 
-- Create a table
CREATE TABLE users (
    user_id UUID PRIMARY KEY,
    email TEXT,
    full_name TEXT,
    created_at TIMESTAMP,
    last_login TIMESTAMP
);
 
-- Create a table with a 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 operations (within one partition!)
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
 
# Configure execution profile
profile = ExecutionProfile(
    load_balancing_policy=TokenAwarePolicy(
        DCAwareRoundRobinPolicy(local_dc='datacenter1')
    ),
    row_factory=dict_factory  # Return rows as dict
)
 
# Connect to the cluster
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 for performance
insert_user = session.prepare(
    """
    INSERT INTO users (user_id, email, full_name, created_at)
    VALUES (?, ?, ?, toTimestamp(now()))
    """
)
 
# Insert data
user_id = uuid.uuid4()
session.execute(insert_user, (user_id, 'test@example.com', 'Test User'))
 
# Read data
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']}")
 
# Close connection
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()
 
# Initialize Cassandra on startup
cluster = Cluster(['192.168.1.10'])
session = cluster.connect('demo')
session.row_factory = dict_factory
 
# Make the session async
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):
    """Create a 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):
    """Get a user"""
    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])

# Bad: parsing CQL on every request
session.execute(
    f"INSERT INTO users (user_id, email) VALUES ({user_id}, '{email}')"
)
 
# Good: prepare once, reuse many times
insert_stmt = session.prepare(
    "INSERT INTO users (user_id, email, full_name) VALUES (?, ?, ?)"
)
 
# Prepared statement is parsed once, then Cassandra knows the query structure
# Instead of parsing SQL on every insert - just substitute parameters
for user in users:
    session.execute(insert_stmt, (user.id, user.email, user.name))

-- Anti-pattern: trying to emulate JOIN via 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
);
 
-- Problem 1: cannot efficiently get author's articles
-- SELECT * FROM articles WHERE author_id = ? ALLOW FILTERING;
-- ALLOW FILTERING scans ALL partitions - a disaster for performance!
 
-- Problem 2: no sorting by published_at without ALLOW FILTERING

-- Query 1: article by ID
CREATE TABLE articles_by_id (
    article_id UUID PRIMARY KEY,
    author_id UUID,
    title TEXT,
    content TEXT,
    published_at TIMESTAMP
);
 
-- Query 2: author's articles sorted by date
CREATE TABLE articles_by_author (
    author_id UUID,
    published_at TIMESTAMP,
    article_id UUID,
    title TEXT,
    content TEXT,  -- Denormalization: duplicate content
    PRIMARY KEY (author_id, published_at)
) WITH CLUSTERING ORDER BY (published_at DESC);
 
-- Query 3: recent articles (homepage)
CREATE TABLE articles_recent (
    bucket TEXT,  -- For example '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);
 
-- Query 4: comments for an article
CREATE TABLE comments_by_article (
    article_id UUID,
    comment_id TIMEUUID,  -- TIMEUUID for sorting by creation time
    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')
 
# Publish an article - write to THREE tables (denormalization)
article_id = uuid.uuid4()
author_id = uuid.uuid4()
title = "Apache Cassandra: first look"
content = "Expert breakdown of Cassandra..."
published_at = datetime.utcnow()
 
# Table 1: for query by 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)
)
 
# Table 2: for author's articles
session.execute(
    "INSERT INTO articles_by_author (author_id, published_at, article_id, title, content) VALUES (?, ?, ?, ?, ?)",
    (author_id, published_at, article_id, title, content)
)
 
# Table 3: for homepage
session.execute(
    "INSERT INTO articles_recent (bucket, published_at, article_id, author_id, title) VALUES (?, ?, ?, ?, ?)",
    ('home_page', published_at, article_id, author_id, title)
)
 
# Read author's articles - O(1) by partition key
rows = session.execute(
    "SELECT * FROM articles_by_author WHERE author_id = ? LIMIT 20",
    (author_id,)
)
 
# Read comments for an article - O(1) by partition key
rows = session.execute(
    "SELECT * FROM comments_by_article WHERE article_id = ?",
    (article_id,)
)

-- Bad: all events of all users in one partition
CREATE TABLE all_events (
    event_type TEXT,  -- partition key
    event_id TIMEUUID,
    user_id UUID,
    payload TEXT,
    PRIMARY KEY (event_type, event_id)
);
-- Problem: event_type='login' may contain billions of rows -> gigantic partition
 
-- Good: partition per user
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 splits data into millions of small partitions

CREATE TABLE messages (
    chat_id UUID,
    message_id TIMEUUID,  -- Contains creation timestamp
    sender_id UUID,
    text TEXT,
    PRIMARY KEY (chat_id, message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);
 
-- Advantages:
-- 1. Automatic sorting by creation time
-- 2. Uniqueness even with millions of writes per second
-- 3. Can extract timestamp: SELECT dateOf(message_id) FROM messages

-- Insert with TTL (Time To Live) 24 hours
INSERT INTO user_sessions (session_id, user_id, data)
VALUES (uuid(), uuid(), '{"key": "value"}')
USING TTL 86400;
 
-- Cassandra will delete this row in 24 hours automatically
-- No need for cron jobs to clean old data

-- DELETE creates a tombstone
DELETE FROM users WHERE user_id = ?;
 
-- Tombstone remains on disk until compaction (10 days by default)
-- If you delete a million rows -> a million tombstones
-- During SELECT Cassandra reads tombstones -> query slowdown

-- Instead of DELETE after 24 hours use TTL
INSERT INTO sessions (session_id, data) VALUES (?, ?)
USING TTL 86400;
 
-- Cassandra deletes on its own in 24 hours, fewer tombstones

-- By default tombstones live 10 days (864000 seconds)
-- If you do not use manual repair, you can shorten to 1 day
ALTER TABLE sessions WITH gc_grace_seconds = 86400;

# cassandra.yaml
tombstone_warn_threshold: 1000 # Warning when 1000 tombstones in a query
tombstone_failure_threshold: 100000 # Error at 100k tombstones

-- Created a table with partition key = user_id
CREATE TABLE users (
    user_id UUID PRIMARY KEY,
    email TEXT,
    full_name TEXT
);
 
-- Want to query by email -> create a secondary index
CREATE INDEX ON users (email);
 
-- The query works
SELECT * FROM users WHERE email = 'test@example.com';

-- Table for queries by email
CREATE TABLE users_by_email (
    email TEXT PRIMARY KEY,
    user_id UUID,
    full_name TEXT
);
 
-- Now query by email is O(1), uses partition key
SELECT * FROM users_by_email WHERE email = 'test@example.com';

-- Error
SELECT * FROM users WHERE full_name = 'John';
-- Error: Cannot execute this query as it might involve data filtering
 
-- Cassandra suggests adding 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)
);

-- Bad partition key: most requests go to one value
CREATE TABLE page_views (
    page_url TEXT,  -- partition key
    view_id TIMEUUID,
    user_id UUID,
    PRIMARY KEY (page_url, view_id)
);
 
-- Problem: homepage '/' gets 90% of the traffic
-- Partition '/' is overloaded while other nodes idle

-- Better: composite partition key
CREATE TABLE page_views (
    page_url TEXT,
    bucket INT,  -- Random number 0-99
    view_id TIMEUUID,
    user_id UUID,
    PRIMARY KEY ((page_url, bucket), view_id)
);
 
-- Now the homepage is spread across 100 partitions:
-- ('/', 0), ('/', 1), ..., ('/', 99)
-- Load is balanced across nodes

import random
 
page_url = '/'
bucket = random.randint(0, 99)  # Random bucket 0-99
 
session.execute(
    "INSERT INTO page_views (page_url, bucket, view_id, user_id) VALUES (?, ?, ?, ?)",
    (page_url, bucket, uuid.uuid1(), user_id)
)

# Read from all 100 buckets in parallel
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]

# Anti-pattern: batch across different partitions
batch = BatchStatement()
for user in users:
    batch.add(insert_stmt, (user.id, user.email, user.name))
session.execute(batch)
 
# Problem: Cassandra has to coordinate writes to all partitions
# Coordinator node becomes a bottleneck

# Correct: parallel async inserts
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)
 
# Wait for all operations to complete
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;
 
-- Both rows have the same partition key (user_id or email)
-- Cassandra guarantees atomicity

# 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 # Log queries longer than 500ms

# cassandra.yaml
authenticator: PasswordAuthenticator # Instead of AllowAllAuthenticator
authorizer: CassandraAuthorizer # Instead of AllowAllAuthorizer
 
# Enable SSL for client-to-node
client_encryption_options:
  enabled: true
  keystore: /path/to/keystore.jks
  keystore_password: changeit
 
# Enable SSL for node-to-node
server_encryption_options:
  internode_encryption: all
  keystore: /path/to/keystore.jks
  keystore_password: changeit

-- Create a role for the app
CREATE ROLE app_user WITH PASSWORD = 'secure_password' AND LOGIN = true;
 
-- Grant rights only to a specific keyspace
GRANT SELECT, MODIFY ON KEYSPACE production_app TO app_user;
 
-- Deny DROP and TRUNCATE
REVOKE DROP ON KEYSPACE production_app FROM app_user;