Real-Time Analytics Engines

Key Characteristics:
                  Druid              Pinot              ClickHouse
Ingestion:      Real-time          Real-time          Real-time + batch
Query latency:  50-500 ms          10-100 ms          1-50 ms (single node)
Scalability:    100s of nodes      1000s of nodes     10s-100s of nodes
Best for:       Dashboards         User-facing        Log analytics
Origin:         Metamarkets        LinkedIn           Yandex
License:        Apache 2.0         Apache 2.0         Apache 2.0

# ============ STEP 1: Install pydruid ============
# pip install pydruid

from pydruid.db import connect
import json
import requests

druid_broker = "http://localhost:8888"

# ============ STEP 2: Create Kafka ingestion spec ============
ingestion_spec = {
    "type": "kafka",
    "spec": {
        "dataSchema": {
            "dataSource": "pageviews",
            "timestampSpec": {
                "column": "timestamp",
                "format": "iso"
            },
            "dimensionsSpec": {
                "dimensions": [
                    "user_id",
                    "page_url",
                    "referrer",
                    "device_type",
                    "country"
                ]
            },
            "metricsSpec": [
                {"type": "count", "name": "count"},
                {"type": "longSum", "name": "page_load_time", "fieldName": "page_load_time"},
                {"type": "hyperUnique", "name": "unique_users", "fieldName": "user_id"}
            ],
            "granularitySpec": {
                "type": "uniform",
                "segmentGranularity": "hour",
                "queryGranularity": "minute",
                "rollup": True  # Pre-aggregate at minute level
            }
        },
        "ioConfig": {
            "topic": "pageviews",
            "consumerProperties": {
                "bootstrap.servers": "localhost:9092"
            },
            "taskCount": 4,  # Parallelism
            "replicas": 1,
            "taskDuration": "PT1H",  # Create new task every hour
            "useEarliestOffset": False
        },
        "tuningConfig": {
            "type": "kafka",
            "maxRowsPerSegment": 5000000
        }
    }
}

# Submit ingestion task
response = requests.post(
    f"{druid_broker}/druid/indexer/v1/supervisor",
    json=ingestion_spec,
    headers={"Content-Type": "application/json"}
)

print(f"✓ Kafka ingestion started: {response.json()}")

"""
Output:
✓ Kafka ingestion started: {'id': 'pageviews'}
Data now streaming from Kafka → Druid with < 1 second latency
"""


# ============ STEP 3: Query Druid with SQL ============
conn = connect(host='localhost', port=8888, path='/druid/v2/sql/', scheme='http')
cursor = conn.cursor()

# Query 1: Real-time pageview counts by country (last 5 minutes)
cursor.execute("""
    SELECT
        country,
        COUNT(*) as pageviews,
        APPROX_COUNT_DISTINCT_DS_HLL(user_id) as unique_users,
        AVG(page_load_time) as avg_load_time
    FROM pageviews
    WHERE __time > CURRENT_TIMESTAMP - INTERVAL '5' MINUTE
    GROUP BY country
    ORDER BY pageviews DESC
    LIMIT 10
""")

print("\n📊 Top Countries (Last 5 minutes):")
for row in cursor.fetchall():
    country, views, users, load_time = row
    print(f"  {country:15} {views:>8,} views  |  {users:>6,} users  |  {load_time:>5.0f}ms")

"""
Output:
📊 Top Countries (Last 5 minutes):
  United States    45,231 views  |  8,912 users  |   342ms
  United Kingdom   12,456 views  |  2,341 users  |   389ms
  Germany           9,876 views  |  1,892 users  |   301ms
  France            7,234 views  |  1,456 users  |   412ms
  Canada            6,123 views  |  1,234 users  |   298ms

Query time: 47ms
"""


# Query 2: Hourly trend (last 24 hours)
# Note: "hour" is a reserved keyword in Druid SQL's parser, so an unquoted
# "as hour" alias fails to parse - use a non-reserved alias instead
cursor.execute("""
    SELECT
        TIME_FLOOR(__time, 'PT1H') as hour_bucket,
        COUNT(*) as pageviews,
        APPROX_COUNT_DISTINCT_DS_HLL(user_id) as unique_users
    FROM pageviews
    WHERE __time > CURRENT_TIMESTAMP - INTERVAL '24' HOUR
    GROUP BY TIME_FLOOR(__time, 'PT1H')
    ORDER BY hour_bucket
""")

print("\n📈 Hourly Pageview Trend:")
for row in cursor.fetchall():
    hour_bucket, views, users = row
    print(f"  {hour_bucket}  {views:>8,} views  ({users:>6,} unique users)")

"""
Output:
📈 Hourly Pageview Trend:
  2024-01-20T00:00:00.000Z    234,567 views  ( 45,231 unique users)
  2024-01-20T01:00:00.000Z    198,432 views  ( 38,912 unique users)
  2024-01-20T02:00:00.000Z    156,789 views  ( 31,234 unique users)
  [... 21 more hours ...]

Note: pydruid returns TIME_FLOOR results as ISO 8601 strings (with T/Z),
not Python datetime objects, since it's a computed SQL expression

Query time: 89ms
"""


# Query 3: Top referrers for specific page (real-time)
cursor.execute("""
    SELECT
        referrer,
        COUNT(*) as visits,
        AVG(page_load_time) as avg_load_time
    FROM pageviews
    WHERE page_url = '/product/laptop-pro'
      AND __time > CURRENT_TIMESTAMP - INTERVAL '10' MINUTE
    GROUP BY referrer
    ORDER BY visits DESC
    LIMIT 5
""")

print("\n🔗 Top Referrers to /product/laptop-pro (Last 10 min):")
for row in cursor.fetchall():
    referrer, visits, load_time = row
    print(f"  {visits:>5} visits from {referrer} (avg {load_time:.0f}ms load time)")

"""
Output:
🔗 Top Referrers to /product/laptop-pro (Last 10 min):
  1,234 visits from google.com (avg 321ms load time)
    892 visits from facebook.com (avg 398ms load time)
    567 visits from twitter.com (avg 276ms load time)
    234 visits from reddit.com (avg 412ms load time)
    123 visits from direct (avg 298ms load time)

Query time: 32ms
"""

# ============ STEP 1: Install Pinot client ============
# pip install pinotdb

from pinotdb import connect

# Connect to Pinot broker
conn = connect(host='localhost', port=8099, path='/query/sql', scheme='http')
cursor = conn.cursor()


# ============ STEP 2: Create table schema ============
# Table configuration for ride-sharing app (like Uber)
table_config = {
    "tableName": "rides",
    "tableType": "REALTIME",
    "segmentsConfig": {
        "timeColumnName": "ride_timestamp",
        "timeType": "MILLISECONDS",
        "replication": 3,
        "retentionTimeUnit": "DAYS",
        "retentionTimeValue": 7
    },
    "tableIndexConfig": {
        "loadMode": "MMAP",
        "invertedIndexColumns": ["driver_id", "rider_id", "city"],
        "noDictionaryColumns": ["ride_id"],
        "starTreeIndexConfigs": [{
            "dimensionsSplitOrder": ["city", "vehicle_type"],
            "skipStarNodeCreationForDimensions": [],
            "functionColumnPairs": [
                "COUNT__*",
                "SUM__fare_amount",
                "MAX__fare_amount"
            ],
            "maxLeafRecords": 10000
        }]
    },
    "ingestionConfig": {
        "streamIngestionConfig": {
            "streamConfigMaps": [{
                "realtime.segment.flush.threshold.rows": 50000,
                "stream.kafka.topic.name": "ride-events",
                "stream.kafka.broker.list": "localhost:9092",
                "stream.kafka.consumer.type": "lowLevel"
            }]
        }
    }
}

# Note: In production, submit this via Pinot Controller REST API
print("✓ Table 'rides' configured with Star-Tree index for fast aggregations")


# ============ STEP 3: Query real-time ride data ============
# Query 1: Current active rides by city (last 5 minutes)
# Note: now() already returns epoch millis (matching ride_timestamp's
# storage type), so wrapping it in ToDateTime() is invalid here -
# ToDateTime() converts millis to a formatted string given a pattern
# argument, it doesn't take a single argument and isn't needed for a
# millis-to-millis comparison
cursor.execute("""
    SELECT
        city,
        COUNT(*) as active_rides,
        COUNT(DISTINCT driver_id) as active_drivers,
        COUNT(DISTINCT rider_id) as active_riders,
        AVG(surge_multiplier) as avg_surge
    FROM rides
    WHERE ride_timestamp > now() - 300000  -- 5 minutes in ms
      AND ride_status = 'in_progress'
    GROUP BY city
    ORDER BY active_rides DESC
    LIMIT 10
""")

print("\n🚗 Active Rides by City (Real-Time):")
for row in cursor:
    city, rides, drivers, riders, surge = row
    print(f"  {city:15} {rides:>5} rides  |  {drivers:>4} drivers  |  {riders:>4} riders  |  {surge:.2f}x surge")

"""
Output:
🚗 Active Rides by City (Real-Time):
  San Francisco   2,341 rides  |  1,892 drivers  |  2,341 riders  |  1.25x surge
  New York        1,987 rides  |  1,543 drivers  |  1,987 riders  |  1.50x surge
  Los Angeles     1,654 rides  |  1,321 drivers  |  1,654 riders  |  1.15x surge
  Chicago           892 rides  |    723 drivers  |    892 riders  |  1.00x surge
  Boston            567 rides  |    456 drivers  |    567 riders  |  1.10x surge

Query latency: 23ms ⚡ (Star-Tree index accelerated the aggregation!)
"""


# Query 2: Driver earnings in last hour (for driver app)
driver_id = 12345

# Note: pinotdb substitutes named %(name)s parameters (Python %-style
# string formatting under the hood), not DB-API "?" positional
# placeholders - passing a tuple for "?" raises AttributeError since it
# expects a dict of named parameters
cursor.execute("""
    SELECT
        COUNT(*) as completed_rides,
        SUM(fare_amount) as total_earnings,
        AVG(fare_amount) as avg_fare,
        AVG(rider_rating) as avg_rating,
        SUM(CASE WHEN surge_multiplier > 1.0 THEN 1 ELSE 0 END) as surge_rides
    FROM rides
    WHERE driver_id = %(driver_id)s
      AND ride_timestamp > now() - 3600000  -- 1 hour
      AND ride_status = 'completed'
""", {"driver_id": driver_id})

print(f"\n💰 Driver {driver_id} Earnings (Last Hour):")
row = cursor.fetchone()
rides, earnings, avg_fare, rating, surge_rides = row
print(f"  Completed: {rides} rides")
print(f"  Earnings: ${earnings:.2f} (avg ${avg_fare:.2f}/ride)")
print(f"  Rating: {rating:.2f} ⭐")
print(f"  Surge rides: {surge_rides}/{rides}")

"""
Output:
💰 Driver 12345 Earnings (Last Hour):
  Completed: 8 rides
  Earnings: $127.50 (avg $15.94/ride)
  Rating: 4.85 ⭐
  Surge rides: 3/8

Query latency: 8ms ⚡ (Pinot's inverted index on driver_id!)
"""


# Query 3: Surge pricing heatmap (for ops dashboard)
cursor.execute("""
    SELECT
        city,
        vehicle_type,
        PERCENTILEEST(surge_multiplier, 50) as median_surge,
        PERCENTILEEST(surge_multiplier, 95) as p95_surge,
        COUNT(*) as rides
    FROM rides
    WHERE ride_timestamp > now() - 600000  -- 10 minutes
    GROUP BY city, vehicle_type
    HAVING COUNT(*) > 10
    ORDER BY p95_surge DESC
""")

print("\n🔥 Surge Pricing Heatmap:")
for row in cursor:
    city, vehicle, median, p95, rides = row
    surge_indicator = "🔴" if p95 > 2.0 else "🟡" if p95 > 1.5 else "🟢"
    print(f"  {surge_indicator} {city:15} {vehicle:10} P50={median:.2f}x  P95={p95:.2f}x  ({rides} rides)")

"""
Output:
🔥 Surge Pricing Heatmap:
  🔴 New York        UberXL     P50=1.8x  P95=2.5x  (234 rides)
  🔴 San Francisco   UberBlack  P50=1.7x  P95=2.3x  (156 rides)
  🟡 Los Angeles     UberX      P50=1.3x  P95=1.8x  (892 rides)
  🟡 Boston          UberXL     P50=1.4x  P95=1.6x  (123 rides)
  🟢 Chicago         UberX      P50=1.1x  P95=1.3x  (567 rides)

Query latency: 18ms ⚡
"""

# Pinot supports upserts (unique to real-time OLAP!)
# Use case: Update order status as it progresses

# Table with upsert enabled (primary key = order_id)
# Note: upsertConfig is a top-level field of the table config, not nested
# under ingestionConfig - nested there it's just an unrecognized property
# that Pinot silently ignores, so upserts never actually turn on. The
# primary key itself is declared in the table's Schema, not in upsertConfig.
upsert_table_config = {
    "tableName": "orders",
    "tableType": "REALTIME",
    "segmentsConfig": {
        "timeColumnName": "order_time",
        "replication": 2
    },
    "tableIndexConfig": {
        "loadMode": "MMAP",
        "invertedIndexColumns": ["user_id", "status"]
    },
    "upsertConfig": {
        "mode": "FULL"  # FULL or PARTIAL upsert
    }
}

# Primary key is declared in the table's schema, not the table config
orders_schema = {
    "schemaName": "orders",
    "primaryKeyColumns": ["order_id"],  # Unique key
    "dimensionFieldSpecs": [
        {"name": "order_id", "dataType": "STRING"},
        {"name": "status", "dataType": "STRING"}
    ],
    "dateTimeFieldSpecs": [
        {"name": "order_time", "dataType": "LONG", "format": "1:MILLISECONDS:EPOCH", "granularity": "1:MILLISECONDS"}
    ]
}

# Stream of events (same order_id, different status)
# Event 1: {"order_id": 123, "status": "pending", "order_time": 1000}
# Event 2: {"order_id": 123, "status": "confirmed", "order_time": 1050}
# Event 3: {"order_id": 123, "status": "shipped", "order_time": 1100}
# Event 4: {"order_id": 123, "status": "delivered", "order_time": 1200}

# Query always returns LATEST status per order_id
cursor.execute("""
    SELECT order_id, status, order_time
    FROM orders
    WHERE order_id = 123
""")

print("\n📦 Order Status:")
row = cursor.fetchone()
print(f"  Order {row[0]}: {row[1]} (updated at {row[2]})")

"""
Output:
📦 Order Status:
  Order 123: delivered (updated at 1200)

Note: Only the latest record exists in Pinot (old statuses overwritten)
This is MUCH faster than scanning all events to find latest status!
"""

# ============ STEP 1: Install ClickHouse client ============
# pip install clickhouse-driver

from clickhouse_driver import Client
from datetime import datetime, timedelta, timezone
import time

client = Client(host='localhost', port=9000)


# ============ STEP 2: Create table for application logs ============
client.execute("""
    CREATE TABLE IF NOT EXISTS app_logs (
        timestamp DateTime,
        log_level LowCardinality(String),
        service LowCardinality(String),
        message String,
        user_id UInt64,
        request_id String,
        duration_ms UInt32,
        status_code UInt16
    )
    ENGINE = MergeTree()
    PARTITION BY toYYYYMMDD(timestamp)
    ORDER BY (service, timestamp)
    TTL timestamp + INTERVAL 30 DAY  -- Auto-delete after 30 days
""")

print("✓ Table 'app_logs' created with MergeTree engine")


# ============ STEP 3: Insert sample logs (simulating real-time stream) ============
# In production: use Kafka → ClickHouse via Kafka table engine
# Use timezone-aware UTC timestamps: ClickHouse's server clock (and its
# now() in queries below) runs in UTC by default, so a naive datetime.now()
# (local time) would silently insert rows that look hours old/future to the
# server, and every "WHERE timestamp > now() - INTERVAL ..." query below
# would return zero rows outside UTC.
sample_logs = [
    (datetime.now(timezone.utc) - timedelta(seconds=i), 'INFO', 'api-gateway', 'Request received',
     12345 + i, f'req-{i}', 50 + i, 200)
    for i in range(1000)
]

# Add some errors
sample_logs.extend([
    (datetime.now(timezone.utc) - timedelta(seconds=i), 'ERROR', 'payment-service', 'Payment failed',
     20000 + i, f'req-err-{i}', 500 + i, 500)
    for i in range(50)
])

client.execute('INSERT INTO app_logs VALUES', sample_logs)
print(f"✓ Inserted {len(sample_logs)} log records")


# ============ STEP 4: Query logs with sub-second latency ============
# Query 1: Error rate by service (last hour)
result = client.execute("""
    SELECT
        service,
        countIf(log_level = 'ERROR') as errors,
        count() as total_logs,
        round(errors / total_logs * 100, 2) as error_rate_pct
    FROM app_logs
    WHERE timestamp > now() - INTERVAL 1 HOUR
    GROUP BY service
    ORDER BY error_rate_pct DESC
""")

print("\n❌ Error Rates by Service (Last Hour):")
for service, errors, total, error_rate in result:
    indicator = "🔴" if error_rate > 5 else "🟡" if error_rate > 1 else "🟢"
    print(f"  {indicator} {service:20} {errors:>5} errors / {total:>6} logs = {error_rate}%")

"""
Output:
❌ Error Rates by Service (Last Hour):
  🔴 payment-service         50 errors /     50 logs = 100.0%
  🟢 api-gateway              0 errors /   1000 logs = 0.0%

Query time: 12ms (scanned 1,050 rows in 0.012 seconds)

Note: error_rate_pct is computed per service (GROUP BY service), so it's
each service's OWN error rate, not its share of errors across all services.
payment-service in this sample only has ERROR-level records, hence 100%.
"""


# Query 2: 95th percentile latency by endpoint
result = client.execute("""
    SELECT
        service,
        quantile(0.95)(duration_ms) as p95_latency,
        quantile(0.99)(duration_ms) as p99_latency,
        avg(duration_ms) as avg_latency
    FROM app_logs
    WHERE timestamp > now() - INTERVAL 5 MINUTE
    GROUP BY service
""")

print("\n⏱️  Latency Percentiles (Last 5 min):")
for service, p95, p99, avg in result:
    print(f"  {service:20} P95={p95:>4.0f}ms  P99={p99:>4.0f}ms  Avg={avg:>4.0f}ms")

"""
Output:
⏱️  Latency Percentiles (Last 5 min):
  api-gateway          P95= 149ms  P99= 199ms  Avg= 125ms
  payment-service      P95= 549ms  P99= 649ms  Avg= 525ms

Query time: 8ms
"""


# Query 3: Find all logs for specific request_id (troubleshooting)
request_id = 'req-123'

result = client.execute("""
    SELECT
        timestamp,
        log_level,
        service,
        message,
        duration_ms
    FROM app_logs
    WHERE request_id = %(req_id)s
    ORDER BY timestamp
""", {'req_id': request_id})

print(f"\n🔍 Trace for request_id={request_id}:")
for ts, level, service, msg, duration in result:
    print(f"  [{ts}] {level:5} {service:20} {msg} ({duration}ms)")

"""
Output:
🔍 Trace for request_id=req-123:
  [2024-01-20 10:30:45] INFO  api-gateway          Request received (173ms)

Query time: 5ms (ClickHouse's sparse index made this instant!)
"""


# Query 4: Top 10 slowest users (last hour)
result = client.execute("""
    SELECT
        user_id,
        count() as request_count,
        avg(duration_ms) as avg_duration,
        max(duration_ms) as max_duration
    FROM app_logs
    WHERE timestamp > now() - INTERVAL 1 HOUR
      AND user_id != 0
    GROUP BY user_id
    ORDER BY avg_duration DESC
    LIMIT 10
""")

print("\n🐌 Slowest Users (Last Hour):")
for user_id, count, avg_dur, max_dur in result:
    print(f"  User {user_id:>6}: {count:>3} requests, avg {avg_dur:>5.0f}ms, max {max_dur:>5.0f}ms")

"""
Output:
🐌 Slowest Users (Last Hour):
  User  13344:   1 requests, avg  1049ms, max  1049ms
  User  13343:   1 requests, avg  1048ms, max  1048ms
  User  13342:   1 requests, avg  1047ms, max  1047ms
  [... more users ...]

Query time: 15ms

Note: these are api-gateway users, not payment-service - duration_ms in
this sample climbs toward the oldest api-gateway rows (50 + i for i up to
999), which outpaces payment-service's max of ~549ms.
"""


# ============ Performance comparison ============
# Scan 1 billion rows in ClickHouse
client.execute("SELECT count() FROM app_logs")  # Assume 1B rows

start = time.time()
result = client.execute("""
    SELECT count()
    FROM app_logs
    WHERE timestamp > now() - INTERVAL 1 DAY
""")
elapsed = time.time() - start

print(f"\n⚡ ClickHouse Performance:")
print(f"  Scanned: 1,000,000,000 rows")
print(f"  Time: {elapsed:.3f} seconds")
print(f"  Throughput: {1_000_000_000 / elapsed / 1_000_000:.0f} million rows/sec")

"""
Output:
⚡ ClickHouse Performance:
  Scanned: 1,000,000,000 rows
  Time: 0.243 seconds
  Throughput: 4,115 million rows/sec

ClickHouse can scan BILLIONS of rows per second on commodity hardware!
"""

# Create materialized view for common aggregation (service metrics per minute)
# POPULATE backfills the view from rows already in app_logs; without it, a
# materialized view only captures rows inserted AFTER creation and this
# query would return nothing for the data inserted in the previous step.
client.execute("""
    CREATE MATERIALIZED VIEW IF NOT EXISTS service_metrics_mv
    ENGINE = SummingMergeTree()
    PARTITION BY toYYYYMMDD(minute)
    ORDER BY (service, minute)
    POPULATE
    AS SELECT
        toStartOfMinute(timestamp) as minute,
        service,
        count() as request_count,
        sum(duration_ms) as total_duration,
        countIf(status_code >= 500) as error_count
    FROM app_logs
    GROUP BY minute, service
""")

print("✓ Materialized view created: service_metrics_mv")
print("  • Pre-aggregates logs per service per minute")
print("  • Queries hit aggregated data (1000x faster!)")


# Query the materialized view
result = client.execute("""
    SELECT
        service,
        sum(request_count) as requests,
        sum(error_count) as errors,
        round(sum(error_count) / sum(request_count) * 100, 2) as error_rate
    FROM service_metrics_mv
    WHERE minute > now() - INTERVAL 1 HOUR
    GROUP BY service
""")

print("\n📊 Service Metrics (from materialized view):")
for service, requests, errors, error_rate in result:
    print(f"  {service:20} {requests:>6} requests, {errors:>3} errors ({error_rate}%)")

"""
Output:
✓ Materialized view created: service_metrics_mv
  • Pre-aggregates logs per service per minute
  • Queries hit aggregated data (1000x faster!)

📊 Service Metrics (from materialized view):
  api-gateway              1000 requests,   0 errors (0.0%)
  payment-service            50 requests,  50 errors (100.0%)

Query time: 2ms ⚡⚡⚡ (vs 12ms on raw table!)
Materialized views trade storage for speed: perfect for dashboards
"""