ClickHouse Operations: Configuration, Replication, and Production Deployment

<!-- /etc/clickhouse-server/config.d/customhouse>
    <!--.xml -->
<click Network -->
    <listen_host>::</listen_host>
    <http_port>8123</http_port>
    <tcp_port>9000</tcp_port>
    
    <!-- Memory -->
    <max_memory_usage>0</max_memory_usage>  <!-- 0 = unlimited -->
    <max_server_memory_usage>8G</max_server_memory_usage>
    
    <!-- Query -->
    <max_execution_time>300</max_execution_time>
    <max_concurrent_queries>100</max_concurrent_queries>
    
    <!-- Merge tree settings -->
    <max_parts_to_merge_at_once>100</max_parts_to_merge_at_once>
    <max_replicated_fetches_network_bandwidth>100M</max_replicated_fetches_network_bandwidth>
    
    <!-- Compression -->
    <compression>
        <case>
            <method>lz4</method>
        </case>
    </compression>
</clickhouse>

-- Session-level settings
SET max_execution_time = 60;
SET max_memory_usage = '4GB';
SET use_uncompressed_cache = 1;
SET max_threads = 8;

-- Per-user settings
CREATE USER analyst IDENTIFIED BY 'password'
SETTINGS max_execution_time = 120, max_memory_usage = '8GB';

-- Default settings in config
-- /etc/clickhouse-server/users.d/default.xml
<clickhouse>
    <profiles>
        <default>
            <max_memory_usage>16GB</max_memory_usage>
            <use_uncompressed_cache>1</use_uncompressed_cache>
        </default>
    </profiles>
</clickhouse>

<!-- /etc/clickhouse-server/config.d/metrika.xml -->
<clickhouse_remote_servers>
    <cluster_3_shards>
        <shard>
            <replica>
                <host>node1</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <replica>
                <host>node2</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <replica>
                <host>node3</host>
                <port>9000</port>
            </replica>
        </shard>
    </cluster_3_shards>
</clickhouse_remote_servers>

<!-- Zookeeper configuration -->
<zookeeper>
    <node>
        <host>zk1</host>
        <port>2181</port>
    </node>
</zookeeper>

<!-- Macro definitions -->
<macros>
    <shard>01</shard>
    <replica>01</replica>
</macros>

-- Create distributed table
CREATE TABLE events_distributed (
    id UInt64,
    event_name String,
    timestamp DateTime,
    value Float64
) ENGINE = Distributed(
    cluster_3_shards,
    default,
    events_local,
    sipHash64(event_name)
);

-- Or with custom sharding key
CREATE TABLE users_distributed (
    user_id UInt32,
    name String,
    email String
) ENGINE = Distributed(
    cluster_3_shards,
    default,
    users_local,
    user_id  -- Shard by user_id
);

-- Execute on all cluster nodes
SELECT * FROM clusterAllReplicas('cluster_3_shards', default.events_local)
WHERE timestamp >= '2026-01-01';

-- Distributed DDL
CREATE TABLE new_table (
    id UInt64,
    name String
) ENGINE = MergeTree() ORDER BY id
ON CLUSTER cluster_3_shards;

-- Check cluster status
SELECT 
    cluster,
    shard_num,
    replica_num,
    host_name,
    is_active
FROM system.cluster;

-- Create replicated table
CREATE TABLE replicated_events (
    id UInt64,
    event_name String,
    timestamp DateTime,
    value Float64
) ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/events',
    '{replica}',
    ORDER BY (event_name, timestamp),
    PARTITION BY toYYYYMM(timestamp),
    SETTINGS index_granularity = 8192
);

-- With different replica path
CREATE TABLE replicated_logs (
    id UInt64,
    message String,
    timestamp DateTime
) ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/replicated_logs',
    '{replica}',
    ORDER BY id,
    SETTINGS 
        cleanup_delay_period = 60,
        cleanup_delay_period_random_add = 5,
        merge_selecting_policy = 'random'
);

-- Check replication status
SELECT 
    database,
    table,
    is_stale,
    last_exception
FROM system.replicas;

-- Check queue
SELECT 
    database,
    table,
    is_stale,
    last_exception,
    queue_size,
    queue_oldest_time,
    inserts_in_queue,
    merges_in_queue
FROM system.replicas;

-- Check parts
SELECT 
    database,
    table,
    partition,
    sum(rows) as rows,
    sum(bytes) as bytes
FROM system.parts
WHERE active
GROUP BY database, table, partition;

-- Check mutations
SELECT 
    database,
    table,
    mutation_id,
    command,
    create_time,
    apply_time,
    is_done
FROM system.mutations
WHERE is_done = 0;

-- Freeze partition (creates backup reference)
ALTER TABLE events FREEZE PARTITION '202601';

-- List frozen partitions
SELECT * FROM system.frozen_tables;

-- Create backup directory
-- /var/lib/clickhouse/shadow/N/

-- Create backup to S3
BACKUP TABLE events TO S3(
    'https://bucket.s3.amazonaws.com/backup',
    'access_key',
    'secret_key'
)(
    compression_method => 'gzip'
);

-- Restore from S3
RESTORE TABLE events FROM S3(
    'https://bucket.s3.amazonaws.com/backup',
    'access_key',
    'secret_key'
);

-- Incremental backup (using incremental backup feature)
BACKUP TABLE events TO FILE('backup incremental')
SETTINGS base_backup = FILE('backup');

import clickhouse_connect
import subprocess
from datetime import datetime
import os

class ClickHouseBackup:
    def __init__(self, config):
        self.client = clickhouse_connect.get_client(**config)
        self.backup_dir = config.get('backup_dir', '/tmp/clickhouse_backups')
        os.makedirs(self.backup_dir, exist_ok=True)
    
    def create_backup(self, table, partition=None):
        """Create table backup."""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        
        if partition:
            # Freeze specific partition
            self.client.command(f"ALTER TABLE {table} FREEZE PARTITION '{partition}'")
        else:
            # Freeze entire table
            self.client.command(f"ALTER TABLE {table} FREEZE")
        
        print(f"Backup created for {table}")
    
    def restore_backup(self, table, backup_path):
        """Restore table from backup."""
        # Copy files to shadow directory
        # This is simplified - production needs careful handling
        
        # Attach table
        # ATTACH TABLE statements...
        
        print(f"Restored {table}")
    
    def export_to_s3(self, table, s3_path):
        """Export table to S3."""
        # Export to S3 using clickhouse-backup or native S3
        query = f"""
            INSERT INTO TABLE FUNCTION s3(
                '{s3_path}',
                'access_key',
                'secret_key',
                'TSV'
            )
            SELECT * FROM {table}
        """
        self.client.command(query)
        
        print(f"Exported {table} to S3")

# Usage
backup = ClickHouseBackup({
    'host': 'localhost',
    'backup_dir': '/backups'
})

backup.create_backup('events', '202601')

-- Enable parallel replicas
SET max_parallel_replicas = 4;

-- Use materialized views for pre-aggregation
CREATE MATERIALIZED VIEW hourly_stats
ENGINE = SummingMergeTree()
ORDER BY (event_name, hour)
AS SELECT
    event_name,
    toStartOfHour(timestamp) as hour,
    count() as cnt,
    sum(value) as total_value
FROM events
GROUP BY event_name, hour;

-- Use skip index
ALTER TABLE events ADD INDEX idx_name event_name TYPE bloom_filter GRANULARITY 4;

-- Force query plan
EXPLAIN PIPELINE SELECT count() FROM events;

-- Control memory for aggregations
SET max_bytes_before_external_group_by = '4GB';
SET max_bytes_before_external_sort = '4GB';

-- Check memory usage
SELECT 
    query,
    memory_usage,
    read_rows,
    formatReadableSize(memory_usage) as memory
FROM system.query_log
WHERE type = 'QueryFinish'
ORDER BY memory_usage DESC
LIMIT 10;

-- Query with limits
SELECT * FROM huge_table
LIMIT 100
SETTINGS max_block_size = 100;

-- Query performance
SELECT 
    query,
    read_rows,
    read_bytes,
    result_rows,
    result_bytes,
    execution_time,
    memory_usage
FROM system.query_log
WHERE type = 'QueryFinish'
ORDER BY event_time DESC
LIMIT 20;

-- Table sizes
SELECT 
    database,
    table,
    formatReadableSize(sum(bytes)) as size,
    sum(rows) as rows,
    count() as parts
FROM system.parts
WHERE active
GROUP BY database, table
ORDER BY size DESC;

-- Merge operations
SELECT 
    database,
    table,
    partition,
    sum(rows) as rows,
    sum(bytes) as bytes,
    count() as num_parts,
    max(creation_time) as last_merge
FROM system.parts
WHERE level > 0 AND active
GROUP BY database, table, partition
ORDER BY last_merge;

#!/bin/bash
# clickhouse-health.sh

HOST=${1:-localhost}
PORT=${2:-8123}

# Check if ClickHouse is responding
if ! curl -s "http://$HOST:$PORT/ping" > /dev/null; then
    echo "ERROR: ClickHouse not responding"
    exit 1
fi

# Check replication lag
LAG=$(clickhouse-client --query "
    SELECT max(absolute_delay) 
    FROM system.replicas 
    WHERE absolute_delay > 300
")

if [ -n "$LAG" ] && [ "$LAG" -gt 300 ]; then
    echo "WARNING: Replication lag detected: $LAG seconds"
fi

# Check disk space
DISK_USAGE=$(df -h /var/lib/clickhouse | tail -1 | awk '{print $5}' | sed 's/%//')
if [ "$DISK_USAGE" -gt 90 ]; then
    echo "WARNING: Disk usage at ${DISK_USAGE}%"
fi

echo "Health check passed"

<!-- clickhouse-keeper configuration -->
<clickhouse>
    <keeper_server>
        <tcp_port>9181</tcp_port>
        <server_id>1</server_id>
        <log_storage_path>/var/lib/clickhouse/coordination/log</log_storage_path>
        <snapshot_storage_path>/var/lib/clickhouse/coordination/snapshots</snapshot_storage_path>
        
        <raft_configuration>
            <server>
                <id>1</id>
                <hostname>node1</hostname>
                <port>9234</port>
            </server>
            <server>
                <id>2</id>
                <hostname>node2</hostname>
                <port>9234</port>
            </server>
            <server>
                <id>3</id>
                <hostname>node3</hostname>
                <port>9234</port>
            </server>
        </raft_configuration>
    </keeper_server>
</clickhouse>

-- Using random distribution
CREATE DATABASE random_db AS Remote('localhost', 'default', 'events', 'random');

-- Using nearest
CREATE DATABASE nearest_db AS Remote('localhost', 'default', 'events', 'nearest');

-- Using configured settings
CREATE DATABASE weighted_db AS Remote('localhost', 'default', 'events', 
    (shard_weight=2, replica_weight=1));

-- Create user with roles
CREATE USER analyst IDENTIFIED BY 'strong_password'
SETTINGS max_execution_time = 60, max_memory_usage = '4GB';

-- Grant permissions
GRANT SELECT ON default.events TO analyst;
GRANT SELECT ON default.* TO analyst;

-- Create role
CREATE ROLE reporting;
GRANT SELECT ON default.* TO reporting;
GRANT reporting TO analyst;

-- Row-level security
CREATE ROW POLICY user_filter ON events
USING user_id = currentUser();

-- Enable SSL/TLS
-- In config:
<openSSL>
    <server>
        <certificateFile>/path/to/cert.pem</certificateFile>
        <privateKeyFile>/path/to/key.pem</privateKeyFile>
    </server>
</openSSL>

-- Connect with SSL
clickhouse-client --ssl

-- Bad: Many small inserts
INSERT INTO events VALUES (1, 'a');  -- Repeat 1000 times

-- Good: Batch inserts
INSERT INTO events VALUES (1, 'a'), (2, 'b'), ...;  -- Single insert with 1000 rows

-- Recommended batch size: 1000-10000 rows per insert

-- Always partition by time for time-series data
CREATE TABLE events (
    timestamp DateTime,
    ...
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(timestamp)  -- Partition by month
ORDER BY timestamp;

-- Add secondary indexes for filtering
ALTER TABLE events ADD INDEX idx_user user_id TYPE minmax GRANULARITY 4;
ALTER TABLE events ADD INDEX idx_name event_name TYPE bloom_filter GRANULARITY 1;