ffsshhttiikk Apache Cassandra
What I do
I am a highly scalable, distributed NoSQL database designed for handling large amounts of data across many commodity servers without a single point of failure. I use a wide-column store model inspired by Google's Bigtable and operate on a peer-to-peer architecture. I provide high availability, linear scalability, and tunable consistency. I excel at write-heavy workloads, time-series data, and applications requiring global distribution.
When to use me
- Building applications requiring high write throughput (IoT, logging, telemetry)
- Systems needing horizontal scalability across multiple data centers
- Time-series data storage and analysis
- Message queues and event sourcing
- Product catalogs with frequent updates
- Fraud detection and real-time analytics
- Global applications requiring low-latency access from multiple regions
- Applications with predictable growth and high availability requirements
Core Concepts
- Wide-Column Store: Data stored in tables with dynamic columns; rows can have different column sets
- Partition Key: Primary identifier determining data distribution across cluster nodes
- Cluster Key: Combination of partition key and clustering columns determining data ordering
- Consistency Levels: Tunable consistency (ONE, QUORUM, ALL, LOCAL_QUORUM) for read/write operations
- Gossip Protocol: Peer-to-peer communication for cluster membership and state detection
- Merkle Trees: Used for anti-entropy repair to synchronize data between replicas
- Compaction: Process of merging SSTables to optimize read performance and reclaim space
- Tunable Consistency: Balance between consistency and availability based on use case requirements
- Lightweight Transactions: Paxos-based linearizable consistency for conditional operations
- Data Center Awareness: Topology-aware replication and query routing for multi-DC deployments
Code Examples
Basic Connection and CRUD Operations
from cassandra.cluster import Cluster
from cassandra.query import SimpleStatement, ConsistencyLevel
from cassandra import InvalidRequest
cluster = Cluster(["127.0.0.1"], port=9042)
session = cluster.connect("app_keyspace")
def create_user(user_data):
query = """
INSERT INTO users (user_id, email, name, created_at, profile)
VALUES (%s, %s, %s, %s, %s)
USING TTL 2592000
"""
session.execute(query, (
user_data["user_id"],
user_data["email"],
user_data["name"],
user_data["created_at"],
user_data.get("profile", {})
))
def get_user_by_id(user_id):
query = "SELECT * FROM users WHERE user_id = %s"
result = session.execute(query, (user_id,))
return result.one()
def get_user_by_email(email):
query = "SELECT * FROM users_by_email WHERE email = %s"
result = session.execute(query, (email,))
return result.one()
def update_user_profile(user_id, **updates):
set_clauses = []
params = []
for key, value in updates.items():
set_clauses.append(f"{key} = %s")
params.append(value)
params.append(user_id)
query = f"UPDATE users SET {', '.join(set_clauses)} WHERE user_id = %s"
session.execute(query, params)
def delete_user(user_id):
query = "DELETE FROM users WHERE user_id = %s"
session.execute(query, (user_id,))
def add_user_friend(user_id, friend_id, since):
query = """
INSERT INTO user_friends (user_id, friend_id, since)
VALUES (%s, %s, %s)
"""
session.execute(query, (user_id, friend_id, since))
def get_user_friends(user_id):
query = "SELECT friend_id, since FROM user_friends WHERE user_id = %s"
result = session.execute(query, (user_id,))
return [row.friend_id for row in result]
Time-Series Data and Aggregations
from datetime import datetime, timedelta
def write_sensor_reading(sensor_id, reading_type, value, timestamp=None):
if timestamp is None:
timestamp = datetime.utcnow()
query = """
INSERT INTO sensor_readings (
sensor_id, reading_type, timestamp, value, date
) VALUES (%s, %s, %s, %s, %s)
"""
session.execute(query, (sensor_id, reading_type, timestamp, value, timestamp.date()))
def get_sensor_readings(sensor_id, reading_type, start_time, end_time):
query = """
SELECT timestamp, value FROM sensor_readings
WHERE sensor_id = %s AND reading_type = %s
AND timestamp >= %s AND timestamp <= %s
ORDER BY timestamp DESC
"""
result = session.execute(query, (sensor_id, reading_type, start_time, end_time))
return [{"timestamp": row.timestamp, "value": row.value} for row in result]
def get_latest_sensor_readings(sensor_id):
query = """
SELECT reading_type, timestamp, value
FROM latest_sensor_readings
WHERE sensor_id = %s
"""
result = session.execute(query, (sensor_id,))
return {row.reading_type: {"timestamp": row.timestamp, "value": row.value} for row in result}
def update_latest_reading(sensor_id, reading_type, value, timestamp):
query = """
INSERT INTO latest_sensor_readings (sensor_id, reading_type, timestamp, value)
VALUES (%s, %s, %s, %s)
"""
session.execute(query, (sensor_id, reading_type, timestamp, value))
def get_hourly_averages(sensor_id, reading_type, date):
query = """
SELECT hour, avg_value FROM hourly_sensor_stats
WHERE sensor_id = %s AND reading_type = %s AND date = %s
"""
result = session.execute(query, (sensor_id, reading_type, date))
return [{"hour": row.hour, "avg_value": row.avg_value} for row in result]
def get_daily_stats(date):
query = """
SELECT sensor_id, reading_type, min_value, max_value, avg_value, count
FROM daily_sensor_stats
WHERE date = %s
"""
result = session.execute(query, (date,))
return [dict(row._asdict()) for row in result]
Batching and Performance Optimization
from cassandra.query import BatchStatement, BatchType
def batch_insert_orders(orders):
batch = BatchStatement(batch_type=BatchType.UNLOGGED)
for order in orders:
batch.add(SimpleStatement("""
INSERT INTO orders (order_id, user_id, total, status, created_at)
VALUES (%s, %s, %s, %s, %s)
"""), (order["order_id"], order["user_id"], order["total"],
order.get("status", "pending"), order["created_at"]))
for item in order["items"]:
batch.add(SimpleStatement("""
INSERT INTO order_items (order_id, product_id, quantity, price)
VALUES (%s, %s, %s, %s)
"""), (order["order_id"], item["product_id"],
item["quantity"], item["price"]))
session.execute(batch)
def batch_update_inventory(updates):
batch = BatchStatement(batch_type=BatchType.LOGGED)
for update in updates:
batch.add(SimpleStatement("""
UPDATE inventory SET quantity = quantity - %s
WHERE product_id = %s
"""), (update["quantity"], update["product_id"]))
session.execute(batch)
def concurrent_inserts(records, concurrency=50):
from concurrent.futures import ThreadPoolExecutor
def insert_record(record):
session.execute("""
INSERT INTO events (event_id, event_type, data, created_at)
VALUES (%s, %s, %s, %s)
""", (record["event_id"], record["event_type"],
record["data"], record["created_at"]))
with ThreadPoolExecutor(max_workers=concurrency) as executor:
list(executor.map(insert_record, records))
def paginate_results(query, params=None, page_size=1000):
statement = SimpleStatement(query, fetch_size=page_size)
result = session.execute(statement, params) if params else session.execute(statement)
while result.has_more_pages:
for row in result.current_rows:
yield row
result = session.execute(result.next_page())
Materialized Views and Secondary Indexes
def create_materialized_views():
session.execute("""
CREATE MATERIALIZED VIEW users_by_email AS
SELECT * FROM users
WHERE email IS NOT NULL
PRIMARY KEY (email)
""")
session.execute("""
CREATE MATERIALIZED VIEW orders_by_user AS
SELECT * FROM orders
WHERE user_id IS NOT NULL AND order_id IS NOT NULL
PRIMARY KEY (user_id, order_id)
WITH CLUSTERING ORDER BY (order_id DESC)
""")
def get_orders_by_user(user_id, limit=50):
query = "SELECT * FROM orders_by_user WHERE user_id = %s LIMIT %s"
result = session.execute(query, (user_id, limit))
return [dict(row._asdict()) for row in result]
def create_secondary_index(column_name, table_name):
session.execute(f"""
CREATE INDEX idx_{table_name}_{column_name}
ON {table_name} ({column_name})
""")
def search_products_by_category(category):
query = "SELECT * FROM products WHERE category = %s"
result = session.execute(query, (category,))
return [dict(row._asdict()) for row in result]
def get_products_by_multiple_tags(tags):
query = "SELECT * FROM products_by_tag WHERE tag = %s"
results = []
for tag in tags:
result = session.execute(query, (tag,))
results.extend([dict(row._asdict()) for row in result])
return results
Lightweight Transactions and Consistency
from cassandra.query import LightweightTransactionProtocol
def create_user_if_not_exists(user_data):
query = """
INSERT INTO users (user_id, email, name, created_at)
VALUES (%s, %s, %s, %s)
IF NOT EXISTS
"""
result = session.execute(query, (
user_data["user_id"],
user_data["email"],
user_data["name"],
user_data["created_at"]
))
return result.was_applied
def update_with_condition(product_id, new_price, expected_stock):
query = """
UPDATE products SET price = %s
WHERE product_id = %s
IF stock = %s
"""
result = session.execute(query, (new_price, product_id, expected_stock))
return result.was_applied, result[0]
def atomic_counter_increment(counter_id, amount=1):
query = """
UPDATE counters SET count = count + %s WHERE counter_id = %s
"""
statement = SimpleStatement(query, serial_consistency_level=ConsistencyLevel.LOCAL_SERIAL)
result = session.execute(statement, (amount, counter_id))
return result
def read_with_quorum(query, params=None):
statement = SimpleStatement(
query,
consistency_level=ConsistencyLevel.QUORUM
)
result = session.execute(statement, params) if params else session.execute(statement)
return result
def write_with_quorum(data_query, data_params, audit_query, audit_params):
batch = BatchStatement(batch_type=BatchType.LOGGED)
batch.add(SimpleStatement(data_query,
consistency_level=ConsistencyLevel.QUORUM), data_params)
batch.add(SimpleStatement(audit_query,
consistency_level=ConsistencyLevel.QUORUM), audit_params)
session.execute(batch)
Best Practices
- Design Partition Keys Carefully: Partition key determines data distribution; avoid hot partitions with skewed data
- Use Appropriate Consistency Levels: Choose QUORUM for balanced consistency; ONE for maximum availability
- Batch Operations Wisely: Use UNLOGGED batches for idempotent operations; LOGGED for transactional needs
- Avoid Secondary Indexes on High-Cardinality Columns: Secondary indexes perform best on low-cardinality columns
- Use Lightweight Transactions Sparingly: Paxos consensus has performance overhead; use IF NOT EXISTS judiciously
- Implement Proper TTL: Use TTL for automatic data expiration instead of manual deletes where appropriate
- Monitor Compaction: Choose appropriate compaction strategy (STCS, LCS, TWCS) based on workload patterns
- Configure Proper Replication Factor: Use RF=3 for production; consider DC-aware replication strategies
- Use Prepared Statements: Reuse prepared statements to reduce query parsing overhead
- Handle Retries and Timeouts: Implement retry policies and timeout handling for transient failures