Cassandra CQL Cheatsheet: Keys, Consistency and nodetool

# Docker — single-node Cassandra 5 (replace with the latest LTS in prod)
docker run -d --name cass \
    -p 9042:9042 \
    -e CASSANDRA_CLUSTER_NAME=devcluster \
    cassandra:5

# Wait until the node is up (status UN = Up Normal)
docker exec cass nodetool status

# Connect with cqlsh (bundled in the image)
docker exec -it cass cqlsh

# Auth + TLS (typical prod connect)
cqlsh c1.internal 9042 -u app -p **** --ssl

# Bulk import / export from the shell
# Inside cqlsh:
#   COPY app.sensors FROM 'rows.csv' WITH HEADER=true;
#   COPY app.sensors TO   'rows.csv' WITH HEADER=true;

# Python driver
pip install "cassandra-driver>=3.29"

python - <<'PY'
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
auth = PlainTextAuthProvider('cassandra', 'cassandra')
with Cluster(['127.0.0.1'], port=9042, auth_provider=auth) as cluster:
    session = cluster.connect()
    print(session.execute("SELECT release_version FROM system.local").one())
PY

-- Keyspace with NetworkTopologyStrategy (multi-DC ready).
CREATE KEYSPACE IF NOT EXISTS app
WITH replication = {
    'class':           'NetworkTopologyStrategy',
    'dc1':             3,
    'dc2':             3
}
AND durable_writes = true;

USE app;

-- Time-series table: one partition per (sensor_id, day).
-- Partitions stay bounded; reads hit one partition; TWCS handles compaction efficiently.
CREATE TABLE IF NOT EXISTS sensor_readings (
    sensor_id    uuid,
    day          date,
    ts           timestamp,
    temperature  float,
    humidity     float,
    PRIMARY KEY ((sensor_id, day), ts)
)
WITH CLUSTERING ORDER BY (ts DESC)
AND default_time_to_live = 7776000                 -- 90 days
AND gc_grace_seconds     = 86400                   -- 1 day (because TTL data tombstones rapidly)
AND compaction = {
    'class':                  'TimeWindowCompactionStrategy',
    'compaction_window_unit': 'DAYS',
    'compaction_window_size': 1
}
AND compression = { 'sstable_compression': 'LZ4Compressor' };

-- A storage-attached index — works on any column, supports range + collections.
CREATE CUSTOM INDEX IF NOT EXISTS sensor_temp_sai
ON sensor_readings (temperature)
USING 'StorageAttachedIndex';

-- Lightweight transactions (LWT). Use sparingly — every LWT is a Paxos round-trip.

-- Insert only if the row is new (claim a username)
INSERT INTO accounts (username, user_id, created_at)
VALUES ('ada', uuid(), toTimestamp(now()))
IF NOT EXISTS;
-- Returns: [applied] = true, or [applied] = false with the existing row.

-- Conditional update — classic compare-and-set
UPDATE accounts
SET    email = 'ada@example.com'
WHERE  username = 'ada'
IF     email   = 'old@example.com';

-- Multi-column predicate
UPDATE counters SET balance = 90, version = 2
WHERE  account_id = 7
IF     balance = 100 AND version = 1;

-- Reading what an LWT just decided — needs SERIAL / LOCAL_SERIAL consistency
SELECT email FROM accounts
WHERE  username = 'ada'
USING  CONSISTENCY LOCAL_SERIAL;

from cassandra.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT
from cassandra.auth import PlainTextAuthProvider
from cassandra.policies import DCAwareRoundRobinPolicy, TokenAwarePolicy
from cassandra import ConsistencyLevel
from cassandra.query import BatchStatement, BatchType
from uuid import uuid4
from datetime import datetime, date

# Connection profile: token-aware → DC-aware. Default CL = LOCAL_QUORUM.
profile = ExecutionProfile(
    load_balancing_policy = TokenAwarePolicy(DCAwareRoundRobinPolicy(local_dc='dc1')),
    consistency_level     = ConsistencyLevel.LOCAL_QUORUM,
    request_timeout       = 10.0,
)

cluster = Cluster(
    contact_points = ['c1.internal', 'c2.internal'],
    port           = 9042,
    auth_provider  = PlainTextAuthProvider('app', '****'),
    execution_profiles = {EXEC_PROFILE_DEFAULT: profile},
)
session = cluster.connect('app')

# Prepared statements — cached on every coordinator, immune to injection.
insert = session.prepare("""
    INSERT INTO sensor_readings (sensor_id, day, ts, temperature, humidity)
    VALUES (?, ?, ?, ?, ?)
""")

# Logged batch only for same-partition writes; otherwise prefer concurrent execute.
batch = BatchStatement(batch_type=BatchType.LOGGED)
sid, today, now = uuid4(), date.today(), datetime.utcnow()
batch.add(insert, (sid, today, now, 22.4, 41.2))
session.execute(batch)

# Async fanout
futures = [session.execute_async(insert, (sid, today, now, t, h)) for t, h in samples]
for f in futures: f.result()                              # block / surface errors

# End-to-end: time-series sensor ingest + per-day query, Python driver.
from cassandra.cluster import Cluster
from cassandra import ConsistencyLevel
from uuid import uuid4
from datetime import datetime, date, timedelta
import random

with Cluster(['127.0.0.1']) as cluster:
    s = cluster.connect()

    s.execute("""
        CREATE KEYSPACE IF NOT EXISTS demo
        WITH replication = {'class':'SimpleStrategy','replication_factor':1}
    """)
    s.execute("USE demo")
    s.execute("""
        CREATE TABLE IF NOT EXISTS sensor_readings (
            sensor_id uuid, day date, ts timestamp, temperature float,
            PRIMARY KEY ((sensor_id, day), ts)
        ) WITH CLUSTERING ORDER BY (ts DESC)
          AND default_time_to_live = 86400
          AND compaction = {'class':'TimeWindowCompactionStrategy',
                            'compaction_window_unit':'DAYS','compaction_window_size':1}
    """)

    insert = s.prepare("INSERT INTO sensor_readings (sensor_id, day, ts, temperature) "
                       "VALUES (?, ?, ?, ?)")
    select = s.prepare("SELECT ts, temperature FROM sensor_readings "
                       "WHERE sensor_id = ? AND day = ? LIMIT 5")

    sid, today = uuid4(), date.today()
    for i in range(20):
        s.execute(insert, (sid, today, datetime.utcnow() - timedelta(seconds=i),
                           20 + random.random() * 5),
                  execution_profile=None)
    for row in s.execute(select, (sid, today)):
        print(row.ts, round(row.temperature, 2))