Cassandra in Production: Real-World Patterns and Best Practices

Introduction

Cassandra powers production systems across virtually every industry requiring high write throughput and global distribution. This article explores real-world use cases with practical implementation patterns.

IoT Platforms

Time-Series Sensor Data

-- Sensor readings table
CREATE TABLE sensor_readings (
    device_id TEXT,
    timestamp TIMESTAMP,
    temperature DECIMAL,
    humidity DECIMAL,
    battery_level DECIMAL,
    signal_strength INT,
    PRIMARY KEY (device_id, timestamp)
) WITH CLUSTERING ORDER BY (timestamp DESC)
AND compaction = {
    'class': 'TimeWindowCompactionStrategy',
    'compaction_window_unit': 'HOURS',
    'compaction_window_size': 24
};

-- Device metadata
CREATE TABLE device_metadata (
    device_id TEXT PRIMARY KEY,
    device_type TEXT,
    firmware_version TEXT,
    location TEXT,
    last_seen TIMESTAMP,
    status TEXT
);

-- Alerts
CREATE TABLE device_alerts (
    device_id TEXT,
    alert_id TIMEUUID,
    timestamp TIMESTAMP,
    alert_type TEXT,
    severity TEXT,
    message TEXT,
    acknowledged BOOLEAN DEFAULT false,
    PRIMARY KEY ((device_id), alert_id)
) WITH CLUSTERING ORDER BY (alert_id DESC);

IoT Python Application

from cassandra.cluster import Cluster
from datetime import datetime
import json

class IoTPlatform:
    def __init__(self):
        self.cluster = Cluster(['cassandra-node1', 'cassandra-node2'])
        self.session = self.cluster.connect('iot_platform')
    
    def ingest_sensor_data(self, device_id, reading):
        """Ingest sensor reading"""
        self.session.execute(
            """
            INSERT INTO sensor_readings 
            (device_id, timestamp, temperature, humidity, battery_level, signal_strength)
            VALUES (%s, %s, %s, %s, %s, %s)
            """,
            (device_id, datetime.now(), 
             reading['temperature'], reading['humidity'],
             reading['battery'], reading['signal'])
        )
        
        # Update device metadata
        self.session.execute(
            """
            INSERT INTO device_metadata 
            (device_id, last_seen, status)
            VALUES (%s, %s, 'active')
            """,
            (device_id, datetime.now())
        )
    
    def get_device_readings(self, device_id, hours=24):
        """Get recent readings"""
        rows = self.session.execute(
            """
            SELECT * FROM sensor_readings 
            WHERE device_id = %s 
            AND timestamp > now() - %dh
            """,
            (device_id, hours)
        )
        return list(rows)
    
    def create_alert(self, device_id, alert_type, severity, message):
        """Create alert"""
        self.session.execute(
            """
            INSERT INTO device_alerts 
            (device_id, alert_id, timestamp, alert_type, severity, message)
            VALUES (%s, now(), now(), %s, %s, %s)
            """,
            (device_id, alert_type, severity, message)
        )

Messaging Systems

Message Storage

-- Messages table
CREATE TABLE messages (
    message_id TIMEUUID,
    conversation_id UUID,
    sender_id UUID,
    recipient_id UUID,
    content TEXT,
    created_at TIMESTAMP,
    read_at TIMESTAMP,
    deleted_at TIMESTAMP,
    PRIMARY KEY (conversation_id, message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);

-- Conversations
CREATE TABLE conversations (
    conversation_id UUID PRIMARY KEY,
    participants SET<UUID>,
    last_message_at TIMESTAMP,
    last_message_preview TEXT,
    unread_count MAP<UUID, INT>
);

-- User conversations
CREATE TABLE user_conversations (
    user_id UUID,
    conversation_id UUID,
    last_read_message_id TIMEUUID,
    added_at TIMESTAMP,
    PRIMARY KEY (user_id, last_message_at)
) WITH CLUSTERING ORDER BY (last_message_at DESC);

Message Operations

class MessagingPlatform:
    def __init__(self):
        self.cluster = Cluster(['cassandra-node1', 'cassandra-node2'])
        self.session = self.cluster.connect('messaging')
    
    def send_message(self, sender_id, recipient_id, content):
        """Send a message"""
        import uuid
        from datetime import datetime
        
        conversation_id = uuid.uuid4()
        message_id = uuid.uuid1()
        
        # Create conversation if needed
        self.session.execute(
            """
            INSERT INTO conversations 
            (conversation_id, participants, last_message_at, last_message_preview)
            VALUES (%s, {%s, %s}, %s, %s)
            """,
            (conversation_id, sender_id, recipient_id, 
             datetime.now(), content[:50])
        )
        
        # Insert message
        self.session.execute(
            """
            INSERT INTO messages 
            (message_id, conversation_id, sender_id, recipient_id, content, created_at)
            VALUES (%s, %s, %s, %s, %s, %s)
            """,
            (message_id, conversation_id, sender_id, recipient_id, 
             content, datetime.now())
        )
        
        return conversation_id
    
    def get_conversation(self, conversation_id, limit=50):
        """Get conversation messages"""
        rows = self.session.execute(
            """
            SELECT * FROM messages 
            WHERE conversation_id = %s
            LIMIT %s
            """,
            (conversation_id, limit)
        )
        return list(rows)

User Activity Tracking

Event Logging

-- User events
CREATE TABLE user_events (
    user_id UUID,
    event_type TEXT,
    event_id TIMEUUID,
    timestamp TIMESTAMP,
    properties MAP<TEXT, TEXT>,
    session_id UUID,
    PRIMARY KEY ((user_id, event_type), event_id)
) WITH CLUSTERING ORDER BY (event_id DESC);

-- Daily active users
CREATE TABLE daily_active_users (
    date DATE,
    user_id UUID,
    first_seen TIMESTAMP,
    last_seen TIMESTAMP,
    event_count INT,
    PRIMARY KEY (date, user_id)
);

-- Event aggregations
CREATE TABLE event_aggregations (
    event_type TEXT,
    date DATE,
    count COUNTER,
    unique_users COUNTER,
    PRIMARY KEY ((event_type), date)
);

Analytics Pipeline

class AnalyticsTracker:
    def __init__(self):
        self.cluster = Cluster(['cassandra-node1', 'cassandra-node2'])
        self.session = self.cluster.connect('analytics')
    
    def track_event(self, user_id, event_type, properties=None, session_id=None):
        """Track user event"""
        import json
        from datetime import datetime
        
        # Insert event
        self.session.execute(
            """
            INSERT INTO user_events 
            (user_id, event_type, event_id, timestamp, properties, session_id)
            VALUES (%, event_type%, %, %, %, %)
            """,
            (user_id, event_type, datetime.now(), 
             json.dumps(properties) if properties else None,
             session_id)
        )
    
    def get_user_journey(self, user_id, event_type=None, limit=100):
        """Get user event history"""
        if event_type:
            rows = self.session.execute(
                """
                SELECT * FROM user_events 
                WHERE user_id = %s AND event_type = %s
                LIMIT %s
                """,
                (user_id, event_type, limit)
            )
        else:
            rows = self.session.execute(
                """
                SELECT * FROM user_events 
                WHERE user_id = %s
                LIMIT %s
                """,
                (user_id, limit)
            )
        return list(rows)

Gaming Applications

Player Data

-- Player profiles
CREATE TABLE player_profiles (
    player_id UUID PRIMARY KEY,
    username TEXT,
    email TEXT,
    level INT,
    experience BIGINT,
    currency MAP<TEXT, INT>,
    created_at TIMESTAMP,
    last_login TIMESTAMP
);

-- Player inventory
CREATE TABLE player_inventory (
    player_id UUID,
    item_id UUID,
    quantity INT,
    acquired_at TIMESTAMP,
    PRIMARY KEY ((player_id), item_id)
);

-- Player scores
CREATE TABLE player_scores (
    game_id TEXT,
    player_id UUID,
    score INT,
    achieved_at TIMESTAMP,
    PRIMARY KEY ((game_id), score, player_id)
) WITH CLUSTERING ORDER BY (score DESC);

-- Leaderboards
CREATE TABLE leaderboard_daily (
    game_id TEXT,
    date DATE,
    player_id UUID,
    score INT,
    rank INT,
    PRIMARY KEY ((game_id, date), rank)
);

Game Server

class GameBackend:
    def __init__(self):
        self.cluster = Cluster(['cassandra-node1', 'cassandra-node2'])
        self.session = self.cluster.connect('gaming')
    
    def record_score(self, game_id, player_id, score):
        """Record player score"""
        from datetime import datetime
        
        self.session.execute(
            """
            INSERT INTO player_scores 
            (game, score, achieved_at)
            VALUES (%s, %s, %s, %s)
            """,
           _id, player_id (game_id, player_id, score, datetime.now())
        )
    
    def get_leaderboard(self, game_id, limit=10):
        """Get top scores"""
        rows = self.session.execute(
            """
            SELECT * FROM player_scores 
            WHERE game_id = %s
            LIMIT %s
            """,
            (game_id, limit)
        )
        return list(rows)
    
    def update_inventory(self, player_id, item_id, quantity):
        """Update player inventory"""
        self.session.execute(
            """
            UPDATE player_inventory 
            SET quantity = quantity + %s
            WHERE player_id = %s AND item_id = %s
            """,
            (quantity, player_id, item_id)
        )

Financial Applications

Transaction Logging

-- Transactions
CREATE TABLE transactions (
    transaction_id TIMEUUID,
    account_id UUID,
    transaction_type TEXT,
    amount DECIMAL,
    balance_after DECIMAL,
    timestamp TIMESTAMP,
    status TEXT,
    metadata MAP<TEXT, TEXT>,
    PRIMARY KEY (account_id, timestamp, transaction_id)
) WITH CLUSTERING ORDER BY (timestamp DESC);

-- Account balances
CREATE TABLE account_balances (
    account_id UUID PRIMARY KEY,
    balance DECIMAL,
    currency TEXT,
    last_transaction_id TIMEUUID,
    updated_at TIMESTAMP
);

-- Transaction history
CREATE TABLE transaction_history (
    account_id UUID,
    year INT,
    month INT,
    transaction_id TIMEUUID,
    amount DECIMAL,
    PRIMARY KEY ((account_id, year, month), transaction_id)
) WITH CLUSTERING ORDER BY (transaction_id DESC);

Best Practices Summary

Use Case	Key Features	Cassandra Solution
IoT	High write throughput	TWCS compaction
Messaging	Low latency reads	Time clustering
Gaming	Real-time scores	Counter tables
Analytics	Aggregation	Materialized views
Financial	Audit trail	Time partitioning

Performance Optimization

Write-Heavy Workloads

-- Batch non-logged for same partition
BEGIN BATCH
INSERT INTO events (user_id, event) VALUES (1, 'event1');
INSERT INTO events (user_id, event) VALUES (1, 'event2');
APPLY BATCH;

-- Avoid batch for different partitions
-- Use asynchronous writes instead

Read Optimization

-- Denormalize for read patterns
-- Create separate tables for queries
-- Use appropriate clustering order
-- Limit partition size

Conclusion

Cassandra excels in production environments requiring high write throughput, linear scalability, and fault tolerance. From IoT platforms to gaming leaderboards, Cassandra powers real-time applications across industries.

With proper data modeling and cluster management, Cassandra can handle millions of writes per second while maintaining low latency reads.

Cassandra in Production: Real-World Patterns and Best Practices

Introduction

IoT Platforms

Time-Series Sensor Data

IoT Python Application

Messaging Systems

Message Storage

Message Operations

User Activity Tracking

Event Logging

Analytics Pipeline

Gaming Applications

Player Data

Game Server

Financial Applications

Transaction Logging

Best Practices Summary

Performance Optimization

Write-Heavy Workloads

Read Optimization

Conclusion

Resources

Comments

Share this article

👍 Was this article helpful?