ClickHouse for Real-Time Analytics: Complete Implementation Guide

-- ClickHouse performance characteristics
SELECT 
    'Columnar Storage' as feature,
    'Optimized for analytical queries' as benefit
UNION ALL
SELECT 'Vectorized Execution', 'SIMD operations for speed'
UNION ALL  
SELECT 'Data Compression', '10x better compression than row stores'
UNION ALL
SELECT 'Horizontal Scaling', 'Distributed queries across clusters'
UNION ALL
SELECT 'SQL Compatibility', 'Standard SQL with extensions'
UNION ALL
SELECT 'Real-time Ingestion', 'Millions of rows per second'

# Create ClickHouse directory structure
mkdir -p ~/clickhouse/{data,logs,config}
cd ~/clickhouse

# Create docker-compose.yml
cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
  clickhouse-server:
    image: clickhouse/clickhouse-server:latest
    container_name: clickhouse-server
    ports:
      - "8123:8123"   # HTTP interface
      - "9000:9000"   # Native interface
      - "9009:9009"   # HTTP interface for inter-server communication
    volumes:
      - ./data:/var/lib/clickhouse
      - ./logs:/var/log/clickhouse-server
      - ./config:/etc/clickhouse-server/config.d
    environment:
      CLICKHOUSE_DB: analytics
      CLICKHOUSE_USER: analyst
      CLICKHOUSE_PASSWORD: secure_password
    ulimits:
      nofile:
        soft: 262144
        hard: 262144
    restart: unless-stopped

  clickhouse-client:
    image: clickhouse/clickhouse-client:latest
    container_name: clickhouse-client
    depends_on:
      - clickhouse-server
    command: --host clickhouse-server
    profiles:
      - client

volumes:
  clickhouse_data:
EOF

# Start ClickHouse
docker-compose up -d

# Test connection
docker-compose run --rm clickhouse-client --query "SELECT version()"

<!-- /etc/clickhouse-server/config.d/production.xml -->
<clickhouse>
    <logger>
        <level>information</level>
        <log>/var/log/clickhouse-server/clickhouse-server.log</log>
        <errorlog>/var/log/clickhouse-server/clickhouse-server.err.log</errorlog>
        <size>1000M</size>
        <count>10</count>
    </logger>

    <!-- Memory management -->
    <max_memory_usage>10000000000</max_memory_usage>
    <max_server_memory_usage>25000000000</max_server_memory_usage>
    
    <!-- Performance settings -->
    <max_concurrent_queries>100</max_concurrent_queries>
    <max_thread_pool_size>10000</max_thread_pool_size>
    
    <!-- Compression -->
    <compression>
        <case>
            <min_part_size>10000000000</min_part_size>
            <min_part_size_ratio>0.01</min_part_size_ratio>
            <method>lz4hc</method>
            <level>9</level>
        </case>
    </compression>
    
    <!-- Merge tree settings -->
    <merge_tree>
        <max_suspicious_broken_parts>5</max_suspicious_broken_parts>
        <parts_to_throw_insert>3000</parts_to_throw_insert>
        <parts_to_delay_insert>1000</parts_to_delay_insert>
        <inactive_parts_to_throw_insert>0</inactive_parts_to_throw_insert>
    </merge_tree>

    <!-- Distributed settings -->
    <distributed_ddl>
        <path>/clickhouse/task_queue/ddl</path>
    </distributed_ddl>
</clickhouse>

-- Create users and roles
CREATE USER analyst_read IDENTIFIED WITH plaintext_password BY 'read_password';
CREATE USER analyst_write IDENTIFIED WITH plaintext_password BY 'write_password';
CREATE USER admin_user IDENTIFIED WITH plaintext_password BY 'admin_password';

-- Create roles
CREATE ROLE analyst_role;
CREATE ROLE admin_role;

-- Grant permissions
GRANT SELECT ON analytics.* TO analyst_role;
GRANT INSERT, SELECT ON analytics.* TO analyst_write;
GRANT ALL ON *.* TO admin_role;

-- Assign roles to users
GRANT analyst_role TO analyst_read;
GRANT analyst_role TO analyst_write;
GRANT admin_role TO admin_user;

-- Enable row-level security (if needed)
CREATE ROW POLICY tenant_isolation ON analytics.events
FOR SELECT USING tenant_id = currentUser()
TO analyst_role;

-- Web analytics events table
CREATE TABLE analytics.web_events
(
    timestamp DateTime64(3) CODEC(Delta, LZ4),
    date Date MATERIALIZED toDate(timestamp),
    user_id String CODEC(LZ4),
    session_id String CODEC(LZ4),
    event_type LowCardinality(String),
    page_url String CODEC(LZ4),
    referrer String CODEC(LZ4),
    user_agent String CODEC(LZ4),
    ip_address IPv4 CODEC(LZ4),
    country LowCardinality(String),
    city String CODEC(LZ4),
    device_type LowCardinality(String),
    browser LowCardinality(String),
    
    -- Custom properties as JSON
    properties String CODEC(LZ4),
    
    -- Computed columns for faster queries
    hour UInt8 MATERIALIZED toHour(timestamp),
    day_of_week UInt8 MATERIALIZED toDayOfWeek(timestamp),
    
    -- Performance metrics
    page_load_time UInt32 DEFAULT 0,
    time_on_page UInt32 DEFAULT 0
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, event_type, user_id, timestamp)
SETTINGS index_granularity = 8192;

-- E-commerce transactions table
CREATE TABLE analytics.transactions
(
    transaction_id String CODEC(LZ4),
    timestamp DateTime64(3) CODEC(Delta, LZ4),
    date Date MATERIALIZED toDate(timestamp),
    user_id String CODEC(LZ4),
    session_id String CODEC(LZ4),
    
    -- Transaction details
    total_amount Decimal64(2) CODEC(LZ4),
    currency LowCardinality(String),
    payment_method LowCardinality(String),
    
    -- Location
    country LowCardinality(String),
    region String CODEC(LZ4),
    city String CODEC(LZ4),
    
    -- Items (nested structure)
    items Nested(
        product_id String,
        product_name String,
        category String,
        quantity UInt16,
        unit_price Decimal64(2),
        total_price Decimal64(2)
    ),
    
    -- Marketing attribution
    utm_source String CODEC(LZ4),
    utm_medium String CODEC(LZ4),
    utm_campaign String CODEC(LZ4)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, user_id, timestamp)
SETTINGS index_granularity = 8192;

-- Real-time hourly metrics
CREATE MATERIALIZED VIEW analytics.hourly_metrics
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, hour, event_type, country)
AS SELECT
    date,
    toHour(timestamp) AS hour,
    event_type,
    country,
    count() AS event_count,
    uniq(user_id) AS unique_users,
    uniq(session_id) AS unique_sessions,
    avg(page_load_time) AS avg_page_load_time
FROM analytics.web_events
GROUP BY date, hour, event_type, country;

-- Daily user cohorts
CREATE MATERIALIZED VIEW analytics.daily_cohorts
ENGINE = ReplacingMergeTree()
PARTITION BY toYYYYMM(first_seen_date)
ORDER BY (first_seen_date, user_id)
AS SELECT
    user_id,
    min(date) AS first_seen_date,
    count() AS total_events,
    max(date) AS last_seen_date,
    uniq(session_id) AS total_sessions
FROM analytics.web_events
GROUP BY user_id;

-- Real-time revenue metrics
CREATE MATERIALIZED VIEW analytics.revenue_metrics
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, hour, country, currency)
AS SELECT
    date,
    toHour(timestamp) AS hour,
    country,
    currency,
    count() AS transaction_count,
    sum(total_amount) AS total_revenue,
    avg(total_amount) AS avg_order_value,
    uniq(user_id) AS unique_customers
FROM analytics.transactions
GROUP BY date, hour, country, currency;

-- Using specialized data types for performance
CREATE TABLE analytics.user_behavior
(
    user_id String CODEC(LZ4),
    timestamp DateTime64(3) CODEC(Delta, LZ4),
    
    -- Array columns for multiple values
    viewed_products Array(String) CODEC(LZ4),
    categories_browsed Array(LowCardinality(String)),
    
    -- Map for flexible key-value pairs
    custom_attributes Map(String, String) CODEC(LZ4),
    
    -- Tuple for structured data
    location Tuple(
        latitude Float64,
        longitude Float64,
        accuracy UInt16
    ),
    
    -- Enum for limited value sets
    subscription_tier Enum8('free' = 1, 'premium' = 2, 'enterprise' = 3),
    
    -- Bitmap for efficient set operations
    feature_flags UInt64 CODEC(LZ4),
    
    -- JSON for complex nested structures
    metadata String CODEC(LZ4)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(toDate(timestamp))
ORDER BY (user_id, timestamp)
SETTINGS index_granularity = 8192;

-- Specialized indexes for better performance
ALTER TABLE analytics.user_behavior 
ADD INDEX user_behavior_features_idx feature_flags TYPE bloom_filter GRANULARITY 1;

ALTER TABLE analytics.user_behavior 
ADD INDEX user_behavior_location_idx (location.latitude, location.longitude) TYPE minmax GRANULARITY 1;

-- CSV file ingestion
INSERT INTO analytics.web_events
SELECT
    parseDateTime64BestEffort(timestamp_str) AS timestamp,
    user_id,
    session_id,
    event_type,
    page_url,
    referrer,
    user_agent,
    toIPv4(ip_address_str) AS ip_address,
    country,
    city,
    device_type,
    browser,
    properties,
    toUInt32(page_load_time_str) AS page_load_time,
    toUInt32(time_on_page_str) AS time_on_page
FROM input('timestamp_str String, user_id String, session_id String, event_type String, page_url String, referrer String, user_agent String, ip_address_str String, country String, city String, device_type String, browser String, properties String, page_load_time_str String, time_on_page_str String')
FORMAT CSV;

-- JSON Lines ingestion
INSERT INTO analytics.web_events
SELECT
    timestamp,
    JSONExtractString(json_data, 'user_id') AS user_id,
    JSONExtractString(json_data, 'session_id') AS session_id,
    JSONExtractString(json_data, 'event_type') AS event_type,
    JSONExtractString(json_data, 'page_url') AS page_url,
    JSONExtractString(json_data, 'referrer') AS referrer,
    JSONExtractString(json_data, 'user_agent') AS user_agent,
    toIPv4(JSONExtractString(json_data, 'ip_address')) AS ip_address,
    JSONExtractString(json_data, 'country') AS country,
    JSONExtractString(json_data, 'city') AS city,
    JSONExtractString(json_data, 'device_type') AS device_type,
    JSONExtractString(json_data, 'browser') AS browser,
    JSONExtractString(json_data, 'properties') AS properties,
    JSONExtractUInt(json_data, 'page_load_time') AS page_load_time,
    JSONExtractUInt(json_data, 'time_on_page') AS time_on_page
FROM input('timestamp DateTime64(3), json_data String')
FORMAT JSONEachRow;

# Python client for real-time ingestion
import asyncio
import json
from clickhouse_driver import Client
from datetime import datetime
import uuid

class ClickHouseStreamer:
    def __init__(self, host='localhost', port=9000, database='analytics'):
        self.client = Client(
            host=host,
            port=port,
            database=database,
            settings={'use_numpy': True}
        )
        self.batch_size = 1000
        self.batch_buffer = []
        
    def add_event(self, event_data):
        """Add event to batch buffer"""
        processed_event = self.process_event(event_data)
        self.batch_buffer.append(processed_event)
        
        if len(self.batch_buffer) >= self.batch_size:
            self.flush_batch()
    
    def process_event(self, event_data):
        """Process and validate event data"""
        return {
            'timestamp': datetime.now(),
            'user_id': event_data.get('user_id', ''),
            'session_id': event_data.get('session_id', str(uuid.uuid4())),
            'event_type': event_data.get('event_type', 'unknown'),
            'page_url': event_data.get('page_url', ''),
            'referrer': event_data.get('referrer', ''),
            'user_agent': event_data.get('user_agent', ''),
            'ip_address': event_data.get('ip_address', '0.0.0.0'),
            'country': event_data.get('country', ''),
            'city': event_data.get('city', ''),
            'device_type': event_data.get('device_type', 'unknown'),
            'browser': event_data.get('browser', 'unknown'),
            'properties': json.dumps(event_data.get('properties', {})),
            'page_load_time': event_data.get('page_load_time', 0),
            'time_on_page': event_data.get('time_on_page', 0)
        }
    
    def flush_batch(self):
        """Insert batch to ClickHouse"""
        if not self.batch_buffer:
            return
            
        try:
            self.client.execute(
                'INSERT INTO web_events VALUES',
                self.batch_buffer
            )
            print(f"Inserted {len(self.batch_buffer)} events")
            self.batch_buffer.clear()
        except Exception as e:
            print(f"Error inserting batch: {e}")
            # Implement retry logic or dead letter queue
    
    async def stream_from_kafka(self):
        """Stream events from Kafka"""
        from aiokafka import AIOKafkaConsumer
        
        consumer = AIOKafkaConsumer(
            'web_events',
            bootstrap_servers='localhost:9092',
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
        
        await consumer.start()
        try:
            async for message in consumer:
                self.add_event(message.value)
        finally:
            await consumer.stop()

# Usage example
async def main():
    streamer = ClickHouseStreamer()
    
    # Simulate real-time events
    for i in range(10000):
        event = {
            'user_id': f'user_{i % 1000}',
            'event_type': 'page_view',
            'page_url': f'/page/{i % 100}',
            'country': 'US' if i % 2 == 0 else 'UK',
            'device_type': 'desktop' if i % 3 == 0 else 'mobile'
        }
        streamer.add_event(event)
    
    streamer.flush_batch()  # Flush remaining events

# Run the example
# asyncio.run(main())

-- Create Kafka engine table for real-time ingestion
CREATE TABLE analytics.kafka_events
(
    timestamp DateTime64(3),
    user_id String,
    session_id String,
    event_type String,
    page_url String,
    referrer String,
    user_agent String,
    ip_address String,
    country String,
    city String,
    device_type String,
    browser String,
    properties String,
    page_load_time UInt32,
    time_on_page UInt32
)
ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'localhost:9092',
    kafka_topic_list = 'web_events',
    kafka_group_name = 'clickhouse_analytics',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 4;

-- Create materialized view to move data from Kafka to main table
CREATE MATERIALIZED VIEW analytics.kafka_to_events_mv TO analytics.web_events AS
SELECT
    timestamp,
    user_id,
    session_id,
    event_type,
    page_url,
    referrer,
    user_agent,
    toIPv4OrDefault(ip_address) AS ip_address,
    country,
    city,
    device_type,
    browser,
    properties,
    page_load_time,
    time_on_page
FROM analytics.kafka_events;

-- Analyze query performance
EXPLAIN SYNTAX 
SELECT 
    country,
    count() AS events,
    uniq(user_id) AS unique_users
FROM analytics.web_events 
WHERE date >= '2024-01-01' 
  AND event_type = 'purchase'
GROUP BY country
ORDER BY events DESC;

-- Check query execution plan
EXPLAIN PLAN
SELECT 
    toStartOfHour(timestamp) AS hour,
    count() AS events,
    avg(page_load_time) AS avg_load_time
FROM analytics.web_events 
WHERE date = today()
  AND page_load_time > 0
GROUP BY hour
ORDER BY hour;

-- Analyze table structure and compression
SELECT 
    table,
    name,
    type,
    compression_codec,
    compressed_size,
    uncompressed_size,
    compression_ratio
FROM system.columns 
WHERE database = 'analytics' 
  AND table = 'web_events'
ORDER BY compressed_size DESC;

-- Time series analysis with window functions
SELECT 
    date,
    country,
    events_today,
    events_yesterday,
    events_today - events_yesterday AS change,
    round((events_today - events_yesterday) / events_yesterday * 100, 2) AS change_pct
FROM (
    SELECT 
        date,
        country,
        count() AS events_today,
        neighbor(count(), -1) AS events_yesterday
    FROM analytics.web_events
    WHERE date >= today() - 30
    GROUP BY date, country
    ORDER BY date, country
) 
WHERE events_yesterday > 0;

-- Cohort analysis
WITH user_cohorts AS (
    SELECT 
        user_id,
        toStartOfMonth(min(date)) AS cohort_month,
        min(date) AS first_visit
    FROM analytics.web_events
    GROUP BY user_id
),
cohort_activity AS (
    SELECT 
        c.cohort_month,
        dateDiff('month', c.cohort_month, toStartOfMonth(e.date)) AS period_number,
        count(DISTINCT e.user_id) AS active_users
    FROM user_cohorts c
    JOIN analytics.web_events e ON c.user_id = e.user_id
    GROUP BY cohort_month, period_number
),
cohort_sizes AS (
    SELECT 
        cohort_month,
        count() AS cohort_size
    FROM user_cohorts
    GROUP BY cohort_month
)
SELECT 
    a.cohort_month,
    a.period_number,
    a.active_users,
    s.cohort_size,
    round(a.active_users / s.cohort_size * 100, 2) AS retention_rate
FROM cohort_activity a
JOIN cohort_sizes s ON a.cohort_month = s.cohort_month
ORDER BY a.cohort_month, a.period_number;

-- Funnel analysis
WITH funnel_events AS (
    SELECT 
        user_id,
        session_id,
        event_type,
        timestamp,
        ROW_NUMBER() OVER (PARTITION BY session_id ORDER BY timestamp) AS event_order
    FROM analytics.web_events
    WHERE date = today()
      AND event_type IN ('page_view', 'add_to_cart', 'checkout', 'purchase')
),
funnel_steps AS (
    SELECT 
        session_id,
        countIf(event_type = 'page_view') > 0 AS step1_page_view,
        countIf(event_type = 'add_to_cart') > 0 AS step2_add_to_cart,
        countIf(event_type = 'checkout') > 0 AS step3_checkout,
        countIf(event_type = 'purchase') > 0 AS step4_purchase
    FROM funnel_events
    GROUP BY session_id
)
SELECT 
    'Step 1: Page View' AS step,
    countIf(step1_page_view) AS users,
    round(countIf(step1_page_view) / count() * 100, 2) AS conversion_rate
FROM funnel_steps
UNION ALL
SELECT 
    'Step 2: Add to Cart' AS step,
    countIf(step2_add_to_cart) AS users,
    round(countIf(step2_add_to_cart) / countIf(step1_page_view) * 100, 2) AS conversion_rate
FROM funnel_steps
UNION ALL
SELECT 
    'Step 3: Checkout' AS step,
    countIf(step3_checkout) AS users,
    round(countIf(step3_checkout) / countIf(step2_add_to_cart) * 100, 2) AS conversion_rate
FROM funnel_steps
UNION ALL
SELECT 
    'Step 4: Purchase' AS step,
    countIf(step4_purchase) AS users,
    round(countIf(step4_purchase) / countIf(step3_checkout) * 100, 2) AS conversion_rate
FROM funnel_steps;

-- Use appropriate data types and compression
ALTER TABLE analytics.web_events MODIFY COLUMN event_type LowCardinality(String);
ALTER TABLE analytics.web_events MODIFY COLUMN country LowCardinality(String);

-- Add specialized indexes
ALTER TABLE analytics.web_events 
ADD INDEX user_id_bloom user_id TYPE bloom_filter GRANULARITY 1;

ALTER TABLE analytics.web_events 
ADD INDEX timestamp_minmax timestamp TYPE minmax GRANULARITY 1;

-- Optimize ORDER BY clause for common queries
-- Original table ordered by (date, event_type, user_id, timestamp)
-- Create projection for different access patterns
ALTER TABLE analytics.web_events 
ADD PROJECTION user_activity_projection (
    SELECT 
        user_id,
        timestamp,
        event_type,
        session_id,
        page_url
    ORDER BY user_id, timestamp
);

-- Materialize the projection
ALTER TABLE analytics.web_events MATERIALIZE PROJECTION user_activity_projection;

-- Use dictionaries for dimension data enrichment
CREATE DICTIONARY analytics.country_dict
(
    country_code String,
    country_name String,
    continent String,
    timezone String
)
PRIMARY KEY country_code
SOURCE(HTTP(
    url 'https://raw.githubusercontent.com/haddadi/country-codes/master/country-codes.json'
    format 'JSONEachRow'
))
LIFETIME(MIN 3600 MAX 7200)
LAYOUT(HASHED());

-- Use the dictionary in queries
SELECT 
    dictGet('analytics.country_dict', 'country_name', country) AS country_name,
    dictGet('analytics.country_dict', 'continent', country) AS continent,
    count() AS events
FROM analytics.web_events
WHERE date = today()
GROUP BY country, country_name, continent
ORDER BY events DESC;

-- Real-time metrics for dashboards (updated every few seconds)
-- Current active users (last 5 minutes)
SELECT count(DISTINCT user_id) AS active_users_5min
FROM analytics.web_events
WHERE timestamp >= now() - INTERVAL 5 MINUTE;

-- Current page views per minute
SELECT 
    toStartOfMinute(timestamp) AS minute,
    count() AS page_views
FROM analytics.web_events
WHERE timestamp >= now() - INTERVAL 60 MINUTE
  AND event_type = 'page_view'
GROUP BY minute
ORDER BY minute;

-- Top pages in real-time (last hour)
SELECT 
    page_url,
    count() AS views,
    uniq(user_id) AS unique_viewers,
    avg(page_load_time) AS avg_load_time
FROM analytics.web_events
WHERE timestamp >= now() - INTERVAL 1 HOUR
  AND event_type = 'page_view'
  AND page_load_time > 0
GROUP BY page_url
ORDER BY views DESC
LIMIT 10;

-- Real-time conversion funnel
WITH recent_events AS (
    SELECT 
        session_id,
        event_type,
        timestamp
    FROM analytics.web_events
    WHERE timestamp >= now() - INTERVAL 30 MINUTE
),
session_funnel AS (
    SELECT 
        session_id,
        max(CASE WHEN event_type = 'page_view' THEN 1 ELSE 0 END) AS viewed,
        max(CASE WHEN event_type = 'add_to_cart' THEN 1 ELSE 0 END) AS added_to_cart,
        max(CASE WHEN event_type = 'checkout' THEN 1 ELSE 0 END) AS checkout,
        max(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) AS purchased
    FROM recent_events
    GROUP BY session_id
)
SELECT 
    sum(viewed) AS step1_views,
    sum(added_to_cart) AS step2_cart,
    sum(checkout) AS step3_checkout,
    sum(purchased) AS step4_purchase,
    round(sum(added_to_cart) / sum(viewed) * 100, 2) AS view_to_cart_rate,
    round(sum(checkout) / sum(added_to_cart) * 100, 2) AS cart_to_checkout_rate,
    round(sum(purchased) / sum(checkout) * 100, 2) AS checkout_to_purchase_rate
FROM session_funnel;

-- Anomaly detection using statistical methods
WITH hourly_metrics AS (
    SELECT 
        toStartOfHour(timestamp) AS hour,
        count() AS events,
        uniq(user_id) AS unique_users
    FROM analytics.web_events
    WHERE timestamp >= now() - INTERVAL 7 DAY
    GROUP BY hour
    ORDER BY hour
),
statistics AS (
    SELECT 
        avg(events) AS avg_events,
        stddevPop(events) AS stddev_events,
        avg(unique_users) AS avg_users,
        stddevPop(unique_users) AS stddev_users
    FROM hourly_metrics
    WHERE hour < toStartOfHour(now()) -- Exclude current hour
)
SELECT 
    h.hour,
    h.events,
    s.avg_events,
    h.unique_users,
    s.avg_users,
    abs(h.events - s.avg_events) / s.stddev_events AS events_z_score,
    abs(h.unique_users - s.avg_users) / s.stddev_users AS users_z_score,
    CASE 
        WHEN abs(h.events - s.avg_events) / s.stddev_events > 3 THEN 'ANOMALY'
        WHEN abs(h.events - s.avg_events) / s.stddev_events > 2 THEN 'WARNING'
        ELSE 'NORMAL'
    END AS events_status,
    CASE 
        WHEN abs(h.unique_users - s.avg_users) / s.stddev_users > 3 THEN 'ANOMALY'
        WHEN abs(h.unique_users - s.avg_users) / s.stddev_users > 2 THEN 'WARNING'
        ELSE 'NORMAL'
    END AS users_status
FROM hourly_metrics h
CROSS JOIN statistics s
WHERE h.hour >= now() - INTERVAL 24 HOUR
ORDER BY h.hour DESC;

# Python script for real-time alerting
import time
from clickhouse_driver import Client
import smtplib
from email.mime.text import MIMEText

class ClickHouseAlerting:
    def __init__(self):
        self.client = Client(host='localhost', database='analytics')
        self.alert_thresholds = {
            'low_traffic': 100,  # events per minute
            'high_error_rate': 5.0,  # percentage
            'high_load_time': 5000  # milliseconds
        }
    
    def check_traffic_anomaly(self):
        """Check for traffic anomalies"""
        query = """
        SELECT 
            count() as events_last_minute,
            avg(events_per_minute) as avg_events_per_minute
        FROM (
            SELECT 
                toStartOfMinute(timestamp) as minute,
                count() as events_per_minute
            FROM analytics.web_events
            WHERE timestamp >= now() - INTERVAL 60 MINUTE
            GROUP BY minute
        )
        WHERE minute = (SELECT max(toStartOfMinute(timestamp)) FROM analytics.web_events)
        """
        
        result = self.client.execute(query)
        if result:
            current_events, avg_events = result[0]
            if current_events < self.alert_thresholds['low_traffic']:
                self.send_alert(
                    'Low Traffic Alert',
                    f'Traffic dropped to {current_events} events/min (avg: {avg_events:.0f})'
                )
    
    def check_error_rate(self):
        """Check for high error rates"""
        query = """
        SELECT 
            countIf(JSONExtractString(properties, 'error') != '') as errors,
            count() as total_events,
            errors / total_events * 100 as error_rate
        FROM analytics.web_events
        WHERE timestamp >= now() - INTERVAL 5 MINUTE
        """
        
        result = self.client.execute(query)
        if result:
            errors, total, error_rate = result[0]
            if error_rate > self.alert_thresholds['high_error_rate']:
                self.send_alert(
                    'High Error Rate Alert',
                    f'Error rate: {error_rate:.2f}% ({errors} errors out of {total} events)'
                )
    
    def check_performance(self):
        """Check for performance issues"""
        query = """
        SELECT 
            avg(page_load_time) as avg_load_time,
            quantile(0.95)(page_load_time) as p95_load_time
        FROM analytics.web_events
        WHERE timestamp >= now() - INTERVAL 5 MINUTE
          AND page_load_time > 0
        """
        
        result = self.client.execute(query)
        if result:
            avg_load_time, p95_load_time = result[0]
            if p95_load_time > self.alert_thresholds['high_load_time']:
                self.send_alert(
                    'High Page Load Time Alert',
                    f'P95 load time: {p95_load_time:.0f}ms (avg: {avg_load_time:.0f}ms)'
                )
    
    def send_alert(self, subject, message):
        """Send email alert"""
        print(f"ALERT: {subject} - {message}")
        # Implement actual email sending logic here
    
    def run_monitoring(self):
        """Run continuous monitoring"""
        while True:
            try:
                self.check_traffic_anomaly()
                self.check_error_rate()
                self.check_performance()
                time.sleep(60)  # Check every minute
            except Exception as e:
                print(f"Monitoring error: {e}")
                time.sleep(60)

# Usage
# alerting = ClickHouseAlerting()
# alerting.run_monitoring()

-- Monitor cluster health
SELECT 
    hostname() AS host,
    uptime() AS uptime_seconds,
    version() AS clickhouse_version,
    formatReadableSize(total_memory_tracker) AS memory_usage
FROM system.metrics
WHERE metric = 'MemoryTracking';

-- Check merge performance
SELECT 
    database,
    table,
    count() AS merges_in_progress,
    sum(num_parts) AS parts_being_merged,
    sum(total_size_bytes_compressed) / 1024 / 1024 AS mb_being_merged
FROM system.merges
GROUP BY database, table;

-- Monitor query performance
SELECT 
    query_start_time,
    query_duration_ms,
    read_rows,
    read_bytes,
    memory_usage,
    query
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query_duration_ms > 1000
ORDER BY query_start_time DESC
LIMIT 10;

-- Check part sizes and compression
SELECT 
    database,
    table,
    count() AS parts_count,
    sum(rows) AS total_rows,
    formatReadableSize(sum(data_compressed_bytes)) AS compressed_size,
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed_size,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 2) AS compression_ratio
FROM system.parts
WHERE active = 1
GROUP BY database, table
ORDER BY sum(data_compressed_bytes) DESC;

-- Optimize tables (manually trigger merges)
OPTIMIZE TABLE analytics.web_events FINAL;

-- Drop old partitions (data retention)
ALTER TABLE analytics.web_events 
DROP PARTITION '202312';  -- Drop December 2023 data

-- Analyze table for query optimization
ANALYZE TABLE analytics.web_events;

-- Check and repair table consistency
CHECK TABLE analytics.web_events;

-- Update table statistics
ALTER TABLE analytics.web_events UPDATE STATISTICS;

<!-- Cluster configuration for production -->
<clickhouse>
    <remote_servers>
        <analytics_cluster>
            <shard>
                <replica>
                    <host>ch-node-1</host>
                    <port>9000</port>
                    <user>cluster_user</user>
                    <password>cluster_password</password>
                </replica>
                <replica>
                    <host>ch-node-2</host>
                    <port>9000</port>
                    <user>cluster_user</user>
                    <password>cluster_password</password>
                </replica>
            </shard>
            <shard>
                <replica>
                    <host>ch-node-3</host>
                    <port>9000</port>
                    <user>cluster_user</user>
                    <password>cluster_password</password>
                </replica>
                <replica>
                    <host>ch-node-4</host>
                    <port>9000</port>
                    <user>cluster_user</user>
                    <password>cluster_password</password>
                </replica>
            </shard>
        </analytics_cluster>
    </remote_servers>

    <zookeeper>
        <node>
            <host>zk-1</host>
            <port>2181</port>
        </node>
        <node>
            <host>zk-2</host>
            <port>2181</port>
        </node>
        <node>
            <host>zk-3</host>
            <port>2181</port>
        </node>
    </zookeeper>

    <macros>
        <cluster>analytics_cluster</cluster>
        <shard>01</shard>
        <replica>replica-1</replica>
    </macros>
</clickhouse>

-- Create distributed table for cluster
CREATE TABLE analytics.web_events_distributed AS analytics.web_events
ENGINE = Distributed(analytics_cluster, analytics, web_events, rand());

-- Create replicated table on each node
CREATE TABLE analytics.web_events_local
(
    -- Same schema as web_events
    timestamp DateTime64(3) CODEC(Delta, LZ4),
    date Date MATERIALIZED toDate(timestamp),
    user_id String CODEC(LZ4),
    -- ... other columns
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/web_events', '{replica}')
PARTITION BY toYYYYMM(date)
ORDER BY (date, event_type, user_id, timestamp)
SETTINGS index_granularity = 8192;

#!/bin/bash
# ClickHouse backup script

BACKUP_DIR="/backups/clickhouse/$(date +%Y%m%d)"
CLICKHOUSE_DATA="/var/lib/clickhouse"

# Create backup directory
mkdir -p "$BACKUP_DIR"

# Stop ClickHouse temporarily for consistent backup
sudo systemctl stop clickhouse-server

# Create backup
sudo tar -czf "$BACKUP_DIR/clickhouse_data.tar.gz" -C "$CLICKHOUSE_DATA" .

# Backup metadata
sudo cp -r /etc/clickhouse-server "$BACKUP_DIR/config"

# Start ClickHouse
sudo systemctl start clickhouse-server

# Upload to S3
aws s3 sync "$BACKUP_DIR" "s3://clickhouse-backups/$(date +%Y%m%d)/"

# Cleanup old local backups (keep 7 days)
find /backups/clickhouse -type d -mtime +7 -exec rm -rf {} \;

echo "Backup completed: $BACKUP_DIR"

# Production monitoring script
import time
import json
from clickhouse_driver import Client
from prometheus_client import Gauge, Counter, start_http_server

class ClickHouseMonitoring:
    def __init__(self):
        self.client = Client(host='localhost', database='analytics')
        
        # Prometheus metrics
        self.queries_per_second = Gauge('clickhouse_queries_per_second', 'Queries per second')
        self.memory_usage = Gauge('clickhouse_memory_usage_bytes', 'Memory usage in bytes')
        self.disk_usage = Gauge('clickhouse_disk_usage_bytes', 'Disk usage in bytes')
        self.active_queries = Gauge('clickhouse_active_queries', 'Number of active queries')
        
    def collect_metrics(self):
        """Collect ClickHouse metrics"""
        try:
            # Query performance metrics
            qps_result = self.client.execute("""
                SELECT sum(ProfileEvent_Query) / 60 as qps
                FROM system.events
                WHERE event_time >= now() - INTERVAL 1 MINUTE
            """)
            
            if qps_result:
                self.queries_per_second.set(qps_result[0][0] or 0)
            
            # Memory usage
            memory_result = self.client.execute("""
                SELECT value FROM system.asynchronous_metrics 
                WHERE metric = 'MemoryResident'
            """)
            
            if memory_result:
                self.memory_usage.set(memory_result[0][0])
            
            # Active queries
            active_result = self.client.execute("""
                SELECT count() FROM system.processes
            """)
            
            if active_result:
                self.active_queries.set(active_result[0][0])
                
        except Exception as e:
            print(f"Error collecting metrics: {e}")
    
    def run(self):
        """Run monitoring loop"""
        start_http_server(8000)  # Prometheus metrics endpoint
        
        while True:
            self.collect_metrics()
            time.sleep(30)

# Usage
# monitoring = ClickHouseMonitoring()
# monitoring.run()