Migrating Log Platform from Elasticsearch to ClickHouse: Table Design, Materialized Views & Tuning Lessons

Why Abandon Elasticsearch for ClickHouse

The logging platform handles diverse sources: gateways, microservices, message queues, slow queries, and frontend tracking — over 2 billion logs daily at peak. Query patterns are two-fold: exact lookups by order ID, user ID, or trace_id; and aggregations (success rate, average latency, P99) by API, status code, time window.

Elasticsearch suffered from storage bloat (1 TB raw → 3–5 TB on disk with inverted indexes and replicas) and slow aggregations (group-by scans over billions of documents taking tens of seconds). Logs are write-once, queried mostly within recent days, with stable schemas — a natural fit for ClickHouse’s columnar storage, vectorized execution, and batch inserts. Benchmarks on identical hardware showed ClickHouse aggregation queries an order of magnitude faster, with ~5:1 compression. Druid was evaluated but rejected due to operational complexity.

Spring Boot Integration: Connection Pool & MyBatis

Add the modern JDBC driver and HikariCP:

<dependency>
  <groupId>com.clickhouse</groupId>
  <artifactId>clickhouse-jdbc</artifactId>
  <classifier>all</classifier>
</dependency>
<dependency>
  <groupId>com.zaxxer</groupId>
  <artifactId>HikariCP</artifactId>
</dependency>

Configure a dedicated DataSource in application.yml (pool size 20, min-idle 5, timeout 30s). ClickHouse is not OLTP; a small pool suffices because writes are batched and most queries hit pre-aggregated tables.

spring:
  datasource:
    clickhouse:
      jdbc-url: jdbc:clickhouse://ck-node1:8123,ck-node2:8123/logdb
      driver-class-name: com.clickhouse.jdbc.ClickHouseDriver
      username: default
      password: ${CK_PASSWORD}
      maximum-pool-size: 20
      minimum-idle: 5
      connection-timeout: 30000

Isolate ClickHouse mappers via a separate @MapperScan configuration to avoid mixing with business databases:

@Configuration
@MapperScan(basePackages = "com.youdian.log.mapper.ck",
            sqlSessionFactoryRef = "clickHouseSqlSessionFactory")
public class ClickHouseConfig {
  @Bean("clickHouseDataSource")
  public DataSource clickHouseDataSource() {
    HikariDataSource ds = new HikariDataSource();
    ds.setJdbcUrl("jdbc:clickhouse://ck-node1:8123/logdb");
    ds.setDriverClassName("com.clickhouse.jdbc.ClickHouseDriver");
    ds.setMaximumPoolSize(10);
    return ds;
  }
  @Bean("clickHouseSqlSessionFactory")
  public SqlSessionFactory sqlSessionFactory(@Qualifier("clickHouseDataSource") DataSource ds) throws Exception {
    SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
    bean.setDataSource(ds);
    bean.setTypeAliasesPackage("com.youdian.log.domain");
    return bean.getObject();
  }
}

Mappers use annotated SQL; avoid SELECT * and transactions/updates. Example:

public interface LogStatMapper {
  @Select("SELECT toHour(event_time) AS hour, " +
         "count() AS total, avg(latency) AS avg_latency " +
         "FROM app_log " +
         "WHERE event_date = #{date} AND app_id = #{appId} " +
         "GROUP BY hour ORDER BY hour")
  List<HourlyStat> queryHourlyStat(@Param("date") String date, @Param("appId") String appId);
}

Table Engine & Distributed Table Design

Use ReplicatedMergeTree + Distributed combo. Local table on each shard:

CREATE TABLE logdb.app_log_local (
  event_date Date,
  event_time DateTime,
  app_id String,
  trace_id String,
  user_id UInt64,
  order_id String,
  log_level LowCardinality(String),
  api_path String,
  http_status UInt16,
  latency UInt32,
  message String,
  extra Map(String, String)
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/app_log', '{replica}')
PARTITION BY toYYYYMMDD(event_date)
ORDER BY (event_date, app_id, api_path, event_time)
TTL event_date + INTERVAL 30 DAY DELETE
SETTINGS index_granularity = 8192;

Distributed table:

CREATE TABLE logdb.app_log (
  event_date Date,
  event_time DateTime,
  app_id String,
  trace_id String,
  user_id UInt64,
  order_id String,
  log_level LowCardinality(String),
  api_path String,
  http_status UInt16,
  latency UInt32,
  message String,
  extra Map(String, String)
) ENGINE = Distributed('log_cluster', 'logdb', 'app_log_local', rand());

Design rationale:

Partition by day ( toYYYYMMDD(event_date)) — aligns with query recency and simplifies TTL cleanup. Hourly partitions would create too many parts, slowing merges.

Sort key (event_date, app_id, api_path, event_time) serves the dominant “one day, one app, one API” access pattern. If trace_id lookups are critical, move trace_id earlier (e.g., (event_date, trace_id, event_time)). Wrong sort key forces full scans.

TTL 30 days auto-deletes; for tiered storage use TTL event_date + INTERVAL 30 DAY TO DISK 'cold_volume' (cleanup runs per partition, not real-time). LowCardinality(String) for log_level compresses well; Map(String, String) for extra avoids ALTER TABLE for new fields.

Shard key rand() gives uniform distribution. Using trace_id would co-locate same-trace logs but risks skew.

Replicas vs shards: 3 nodes with 3 shards × 1 replica = no HA. Production needs 3 shards × 2 replicas = 6 nodes. Learned the hard way when a node failed and replica rebuild was painful.

Data Ingestion: Kafka → Spring Boot → ClickHouse

Classic three-stage pipeline: business logs → Kafka (buffer) → Spring Boot consumer → batched ClickHouse inserts.

Consumer config (batch listener, manual ack):

spring:
  kafka:
    consumer:
      max-poll-records: 5000
      enable-auto-commit: false
      auto-offset-reset: latest
      listener:
        type: batch
        ack-mode: manual_immediate

Consumer code parses records, filters nulls, delegates to batch service:

@KafkaListener(topics = "app-log", containerFactory = "kafkaListenerContainerFactory")
public void onMessage(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
  List<LogEntity> list = records.stream()
    .map(r -> parseLog(r.value()))
    .filter(Objects::nonNull)
    .collect(Collectors.toList());
  if (!list.isEmpty()) {
    logBatchService.batchInsert(list);
  }
  ack.acknowledge();
}

Batch insert uses PreparedStatement.addBatch() (1,000–5,000 rows per commit). Single-row inserts are slow; 100k rows caused server memory pressure and merge overhead. 5,000 rows/batch proved optimal.

public void batchInsert(List<LogEntity> logs) {
  String sql = "INSERT INTO app_log (event_date, event_time, app_id, trace_id, user_id, order_id, log_level, api_path, http_status, latency, message, extra) " +
               "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
  try (Connection conn = dataSource.getConnection();
       PreparedStatement ps = conn.prepareStatement(sql)) {
    for (LogEntity log : logs) {
      ps.setDate(1, java.sql.Date.valueOf(log.getEventDate()));
      // ... set other fields with correct types
      ps.addBatch();
    }
    ps.executeBatch();
  }
}

Added bounded ArrayBlockingQueue (100,000 capacity) as async buffer between Kafka consumption and ClickHouse flush (every 5 seconds, drain up to 5,000). Prevents backpressure spikes; larger queues cause GC issues.

@Component
public class LogBatchService {
  private final BlockingQueue<LogEntity> queue = new ArrayBlockingQueue<>(100000);
  @PostConstruct
  public void start() {
    ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    scheduler.scheduleWithFixedDelay(this::flush, 5, 5, TimeUnit.SECONDS);
  }
  public void add(LogEntity log) { queue.offer(log); }
  public void flush() {
    List<LogEntity> batch = new ArrayList<>(5000);
    queue.drainTo(batch, 5000);
    if (batch.isEmpty()) return;
    insertClickHouse(batch);
  }
}

Aggregation Optimization: Materialized Views

High-frequency hourly stats (count, errors, avg/max latency) per app/API would scan billions of rows if run on raw table. Solution: materialized view that incrementally maintains an aggregation table.

Aggregation table uses SummingMergeTree (auto-sums numeric columns on merge):

CREATE TABLE logdb.api_hourly_agg (
  event_date Date,
  hour DateTime,
  app_id String,
  api_path String,
  total_count UInt64,
  error_count UInt64,
  sum_latency UInt64,
  max_latency UInt32
) ENGINE = SummingMergeTree()
PARTITION BY toYYYYMMDD(event_date)
ORDER BY (event_date, hour, app_id, api_path);

Materialized view targets the local table (not distributed) to avoid duplicate triggers:

CREATE MATERIALIZED VIEW logdb.api_hourly_agg_mv TO logdb.api_hourly_agg AS
SELECT
  event_date,
  toStartOfHour(event_time) AS hour,
  app_id,
  api_path,
  count() AS total_count,
  countIf(http_status >= 500) AS error_count,
  sum(latency) AS sum_latency,
  max(latency) AS max_latency
FROM logdb.app_log_local
GROUP BY event_date, hour, app_id, api_path;

Querying the aggregation table returns results in tens of milliseconds. Projection could achieve similar pre-aggregation but requires ALTER TABLE ... ADD PROJECTION on existing large tables (very slow). Materialized views are external, easier to maintain and reason about.

Query Optimization Details

High-cardinality distinct: Avoid COUNT(DISTINCT user_id) (memory heavy, OOM risk). Use uniqCombined(user_id) for ~1% error with controlled memory.

Skip indexes: For columns not in sort key (e.g., user_id, order_id), add minmax index:

ALTER TABLE logdb.app_log_local ADD INDEX idx_user_id user_id TYPE minmax GRANULARITY 2;

Only affects new data; backfill with

ALTER TABLE ... MATERIALIZE INDEX idx_user_id IN PARTITION ...

.

Full-text search: LIKE '%xxx%' needs tokenbf_v1/ngrambf_v1 indexes; not needed for current exact-match/range workloads.

Distributed aggregation network overhead: High group-by cardinality → large intermediate results shuffled across shards. Mitigate by narrowing time range and app_id filters at application layer.

Write Throttling & Server Tuning

During Kafka lag spikes, uncontrolled bulk inserts overwhelm ClickHouse, hurting queries. Added client-side rate limiter (Guava RateLimiter at 50,000 rows/sec):

@Autowired private RateLimiter rateLimiter; // 50000 permits/sec
public void batchInsertWithRateLimit(List<LogEntity> logs) {
  rateLimiter.acquire(logs.size());
  insertClickHouse(logs);
}

Server-side: set max_insert_threads to increase insert parallelism. Avoid mixing multiple partitions in a single insert to reduce merge load.

Slow Query Detection & Data Governance

Monitor system.query_log for queries >3s, alert via scheduled job:

SELECT query, query_duration_ms, read_rows, memory_usage
FROM system.query_log
WHERE event_time > now() - INTERVAL 1 HOUR
  AND query_duration_ms > 3000
ORDER BY query_duration_ms DESC
LIMIT 20;

Governance practices:

Drop health-check/debug logs at consumer.

Handle Kafka duplicate consumption with idempotency; don’t rely on ClickHouse deduplication. ReplacingMergeTree is a last resort (merge overhead).

Keep volatile fields in Map; promote to columns only when stable.

Run periodic OPTIMIZE TABLE to merge small parts.

Capacity Planning & Monitoring

Daily 2B logs × 500 bytes ≈ 1 TB raw. ClickHouse 5:1 compression → 200 GB/day. 30-day retention: 6 TB single replica, 12 TB dual replica. Cluster: 3 nodes, each 8×2 TB NVMe, 64 GB RAM, 16 CPU — currently comfortable headroom. Prometheus + Grafana monitors: node CPU/memory/disk, Kafka consumer lag, ClickHouse query concurrency, slow query count, partition count. Alert thresholds (battle-tested): disk >80%, Kafka lag >1M, avg query latency >2s for 5min, write latency >3s.

Closing Thoughts

System running 1+ years, 50+ apps, 2B+ logs/day, query P99 <200ms, storage cost ~60% lower than Elasticsearch. Key lesson: invest heavily in table design first; scaling and tuning come later. ClickHouse excels when designed right, but punishes poor design severely.