ClickHouse Analytics Automation Script —

ClickHouse คืออะไรและทำไมถึงเร็ว

ClickHouse เป็น open source columnar OLAP database ที่พัฒนาโดย Yandex ออกแบบมาสำหรับ real-time analytical queries บน datasets ขนาดใหญ่ (petabyte-scale) ให้ query speed ที่เร็วกว่า traditional databases หลายร้อยเท่า

เหตุผลที่ ClickHouse เร็วคือ columnar storage อ่านเฉพาะ columns ที่ต้องการ, data compression ที่ลดขนาดข้อมูล 5-10 เท่า, vectorized query execution ที่ process data เป็น batches, parallel processing ใช้ CPU cores ทั้งหมด, primary key index ที่ skip data ไม่ต้องการ และ specialized data structures สำหรับแต่ละ data type

ClickHouse เหมาะสำหรับ log analytics, clickstream analysis, real-time dashboards, time series data, ad-tech analytics, financial data analysis และ IoT data processing ไม่เหมาะสำหรับ transactional workloads (OLTP) ที่ต้อง UPDATE/DELETE บ่อย

เมื่อเทียบกับ alternatives ClickHouse เร็วกว่า PostgreSQL สำหรับ analytical queries 100-1000 เท่า เร็วกว่า Elasticsearch สำหรับ structured data analytics 5-10 เท่า และมี operational cost ต่ำกว่า cloud data warehouses เช่น BigQuery หรือ Snowflake

ติดตั้ง ClickHouse และตั้งค่าเบื้องต้น

วิธีติดตั้งและ configure ClickHouse

# === ติดตั้ง ClickHouse ===



# Ubuntu/Debian

sudo apt-get install -y apt-transport-https ca-certificates curl gnupg

curl -fsSL 'https://packages.clickhouse.com/rpm/lts/repodata/repomd.xml.key' | \

    sudo gpg --dearmor -o /usr/share/keyrings/clickhouse-keyring.gpg

echo "deb [signed-by=/usr/share/keyrings/clickhouse-keyring.gpg] \

    https://packages.clickhouse.com/deb stable main" | \

    sudo tee /etc/apt/sources.list.d/clickhouse.list

sudo apt-get update

sudo apt-get install -y clickhouse-server clickhouse-client



# Start service

sudo systemctl start clickhouse-server

sudo systemctl enable clickhouse-server



# Docker

docker run -d --name clickhouse \

    -p 8123:8123 -p 9000:9000 \

    -v clickhouse-data:/var/lib/clickhouse \

    -v clickhouse-logs:/var/log/clickhouse-server \

    -e CLICKHOUSE_DB=analytics \

    -e CLICKHOUSE_USER=admin \

    -e CLICKHOUSE_PASSWORD=ch_password \

    clickhouse/clickhouse-server:latest



# เชื่อมต่อ

clickhouse-client --user admin --password ch_password



# === สร้าง Database และ Tables ===

CREATE DATABASE IF NOT EXISTS analytics;



-- Web Analytics Events

CREATE TABLE analytics.events (

    event_date Date,

    event_time DateTime64(3),

    user_id UInt64,

    session_id String,

    event_type LowCardinality(String),

    page_url String,

    referrer String,

    country LowCardinality(String),

    device LowCardinality(String),

    browser LowCardinality(String),

    duration_ms UInt32,

    properties Map(String, String)

) ENGINE = MergeTree()

PARTITION BY toYYYYMM(event_date)

ORDER BY (event_date, user_id, event_time)

TTL event_date + INTERVAL 1 YEAR

SETTINGS index_granularity = 8192;



-- Server Metrics

CREATE TABLE analytics.server_metrics (

    timestamp DateTime64(3),

    host LowCardinality(String),

    metric LowCardinality(String),

    value Float64,

    tags Map(String, String)

) ENGINE = MergeTree()

PARTITION BY toYYYYMMDD(timestamp)

ORDER BY (host, metric, timestamp)

TTL timestamp + INTERVAL 90 DAY;



-- Insert sample data

INSERT INTO analytics.events

SELECT

    today() - (rand() % 365) AS event_date,

    toDateTime64(event_date, 3) + rand() % 86400 AS event_time,

    rand() % 100000 AS user_id,

    generateUUIDv4() AS session_id,

    arrayElement(['pageview','click','scroll','purchase'], 1 + rand() % 4) AS event_type,

    concat('/page/', toString(rand() % 50)) AS page_url,

    arrayElement(['google.com','facebook.com','direct','twitter.com'], 1 + rand() % 4) AS referrer,

    arrayElement(['TH','US','JP','SG','DE'], 1 + rand() % 5) AS country,

    arrayElement(['mobile','desktop','tablet'], 1 + rand() % 3) AS device,

    arrayElement(['Chrome','Firefox','Safari','Edge'], 1 + rand() % 4) AS browser,

    rand() % 30000 AS duration_ms,

    map('version', '1.0') AS properties

FROM numbers(10000000);



SELECT count() FROM analytics.events;

SELECT formatReadableSize(total_bytes) FROM system.tables WHERE name = 'events';

สร้าง Analytics Queries และ Materialized Views

Analytical queries และ automated aggregation

# === ClickHouse Analytics Queries ===



-- Daily active users

SELECT

    event_date,

    uniqExact(user_id) AS dau,

    count() AS total_events,

    countIf(event_type = 'purchase') AS purchases,

    round(countIf(event_type = 'purchase') / uniqExact(user_id) * 100, 2) AS conversion_rate

FROM analytics.events

WHERE event_date >= today() - 30

GROUP BY event_date

ORDER BY event_date;



-- Top pages with engagement metrics

SELECT

    page_url,

    count() AS views,

    uniqExact(user_id) AS unique_visitors,

    avg(duration_ms) AS avg_duration_ms,

    quantile(0.50)(duration_ms) AS p50_duration,

    quantile(0.95)(duration_ms) AS p95_duration

FROM analytics.events

WHERE event_type = 'pageview'

  AND event_date >= today() - 7

GROUP BY page_url

ORDER BY views DESC

LIMIT 20;



-- Funnel analysis

SELECT

    level,

    users,

    round(users / first_value(users) OVER (ORDER BY level) * 100, 1) AS pct_of_start

FROM (

    SELECT 1 AS level, uniqExact(user_id) AS users

    FROM analytics.events WHERE event_type = 'pageview' AND event_date = today()

    UNION ALL

    SELECT 2, uniqExact(user_id)

    FROM analytics.events WHERE event_type = 'click' AND event_date = today()

    UNION ALL

    SELECT 3, uniqExact(user_id)

    FROM analytics.events WHERE event_type = 'purchase' AND event_date = today()

)

ORDER BY level;



-- === Materialized Views (auto-aggregation) ===



-- Daily stats materialized view

CREATE MATERIALIZED VIEW analytics.daily_stats_mv

ENGINE = SummingMergeTree()

PARTITION BY toYYYYMM(event_date)

ORDER BY (event_date, country, device)

AS SELECT

    event_date,

    country,

    device,

    count() AS event_count,

    uniqState(user_id) AS unique_users_state,

    sumIf(1, event_type = 'purchase') AS purchase_count,

    avgState(duration_ms) AS avg_duration_state

FROM analytics.events

GROUP BY event_date, country, device;



-- Query materialized view

SELECT

    event_date,

    country,

    sum(event_count) AS events,

    uniqMerge(unique_users_state) AS unique_users,

    sum(purchase_count) AS purchases,

    avgMerge(avg_duration_state) AS avg_duration

FROM analytics.daily_stats_mv

WHERE event_date >= today() - 7

GROUP BY event_date, country

ORDER BY event_date, events DESC;



-- Hourly metrics materialized view

CREATE MATERIALIZED VIEW analytics.hourly_metrics_mv

ENGINE = SummingMergeTree()

ORDER BY (hour, host, metric)

AS SELECT

    toStartOfHour(timestamp) AS hour,

    host,

    metric,

    avg(value) AS avg_value,

    max(value) AS max_value,

    min(value) AS min_value,

    count() AS sample_count

FROM analytics.server_metrics

GROUP BY hour, host, metric;

Automation Scripts ด้วย Python

Python scripts สำหรับ automate ClickHouse tasks

ETL Pipeline และ Data Ingestion

สร้าง ETL pipeline สำหรับ ClickHouse

#!/usr/bin/env python3 # etl_pipeline.py — ClickHouse ETL Pipeline import clickhouse_connect import json import csv import gzip import logging from datetime import datetime from pathlib import Path from typing import List, Dict logging.basicConfig(level=logging.INFO) logger = logging.getLogger("etl") class ClickHouseETL: def __init__(self, client): self.client = client def ingest_csv(self, table, csv_file, batch_size=100000): logger.info(f"Ingesting {csv_file} into {table}") opener = gzip.open if str(csv_file).endswith(".gz") else open with opener(csv_file, "rt") as f: reader = csv.DictReader(f) batch = [] total = 0 columns = None for row in reader: if columns is None: columns = list(row.keys()) batch.append(list(row.values())) if len(batch) >= batch_size: self.client.insert(table, batch, column_names=columns) total += len(batch) batch = [] logger.info(f" Inserted {total} rows") if batch: self.client.insert(table, batch, column_names=columns) total += len(batch) logger.info(f"Total ingested: {total} rows") return total def ingest_json_lines(self, table, jsonl_file, batch_size=50000): logger.info(f"Ingesting {jsonl_file} into {table}") opener = gzip.open if str(jsonl_file).endswith(".gz") else open with opener(jsonl_file, "rt") as f: batch = [] columns = None total = 0 for line in f: record = json.loads(line.strip()) if columns is None: columns = list(record.keys()) batch.append([record.get(c) for c in columns]) if len(batch) >= batch_size: self.client.insert(table, batch, column_names=columns) total += len(batch) batch = [] if batch: self.client.insert(table, batch, column_names=columns) total += len(batch) logger.info(f"Total ingested: {total} rows") return total def transform_and_load(self, source_table, target_table, transform_sql): logger.info(f"Transform: {source_table} -> {target_table}") self.client.command(f""" INSERT INTO {target_table} {transform_sql} """) rows, _ = self.client.query(f"SELECT count() FROM {target_table}") logger.info(f"Target table rows: {rows.result_rows[0][0]}") def run_pipeline(self, steps): logger.info(f"Running pipeline with {len(steps)} steps") start = datetime.now() results = [] for i, step in enumerate(steps): step_start = datetime.now() logger.info(f"Step {i+1}/{len(steps)}: {step['name']}") try: if step["type"] == "ingest_csv": count = self.ingest_csv(step["table"], step["file"]) elif step["type"] == "ingest_jsonl": count = self.ingest_json_lines(step["table"], step["file"]) elif step["type"] == "transform": self.transform_and_load( step["source"], step["target"], step["sql"] ) count = 0 elif step["type"] == "query": self.client.command(step["sql"]) count = 0 results.append({ "step": step["name"], "status": "success", "duration_s": (datetime.now() - step_start).total_seconds(), }) except Exception as e: results.append({ "step": step["name"], "status": "error", "error": str(e), }) logger.error(f"Step failed: {e}") if step.get("required", True): break total_time = (datetime.now() - start).total_seconds() logger.info(f"Pipeline completed in {total_time:.1f}s") return results # client = clickhouse_connect.get_client(host="localhost", username="admin", password="ch_password") # etl = ClickHouseETL(client) # # pipeline = [ # {"name": "Ingest events", "type": "ingest_csv", "table": "analytics.events_raw", "file": "data/events.csv.gz"}, # {"name": "Transform", "type": "transform", "source": "events_raw", "target": "events", # "sql": "SELECT *, toDate(event_time) AS event_date FROM analytics.events_raw"}, # {"name": "Optimize", "type": "query", "sql": "OPTIMI