PostgreSQL 분석 집계 병목, ClickHouse + Node.js 파이프라인으로 최대 100배 빠르게 해소하기

-- 90일 후 삭제
TTL event_time + INTERVAL 90 DAY DELETE
 
-- Hot → Warm → Cold 계층형 이동
TTL event_time + INTERVAL 7 DAY TO DISK 'warm',
    event_time + INTERVAL 30 DAY TO VOLUME 'cold'

-- 원시 이벤트 테이블
CREATE TABLE clickstream_events
(
    event_id    UUID           DEFAULT generateUUIDv4(),
    user_id     UInt64,
    session_id  String,
    event_type  LowCardinality(String),  -- page_view / add_to_cart / purchase
    page_path   String,
    revenue     Decimal(10, 2) DEFAULT 0,
    event_time  DateTime64(3)  DEFAULT now64()
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, page_path, event_time)
TTL event_time + INTERVAL 90 DAY DELETE;

-- MV 집계 결과를 받을 목적지 테이블
CREATE TABLE clickstream_hourly
(
    event_hour    DateTime,
    event_type    LowCardinality(String),
    page_path     String,
    event_count   AggregateFunction(count),
    unique_users  AggregateFunction(uniq, UInt64),
    total_revenue AggregateFunction(sum, Decimal(10, 2))
)
ENGINE = AggregatingMergeTree
ORDER BY (event_hour, event_type, page_path);
 
 
-- Materialized View — 삽입 시점 트리거로 집계
CREATE MATERIALIZED VIEW clickstream_mv
TO clickstream_hourly
AS
SELECT
    toStartOfHour(event_time)  AS event_hour,
    event_type,
    page_path,
    countState()               AS event_count,
    uniqState(user_id)         AS unique_users,
    sumState(revenue)          AS total_revenue
FROM clickstream_events
GROUP BY event_hour, event_type, page_path;

SELECT
    event_hour,
    event_type,
    page_path,
    countMerge(event_count)    AS views,
    uniqMerge(unique_users)    AS unique_users,
    sumMerge(total_revenue)    AS revenue
FROM clickstream_hourly
WHERE event_hour >= now() - INTERVAL 24 HOUR
GROUP BY event_hour, event_type, page_path
ORDER BY event_hour DESC;

<!-- /etc/clickhouse-server/config.d/cluster.xml -->
<clickhouse>
  <remote_servers>
    <events_cluster>
      <shard>
        <replica><host>ch-node-1</host><port>9000</port></replica>
        <replica><host>ch-node-2</host><port>9000</port></replica>
      </shard>
      <shard>
        <replica><host>ch-node-3</host><port>9000</port></replica>
        <replica><host>ch-node-4</host><port>9000</port></replica>
      </shard>
    </events_cluster>
  </remote_servers>
  <macros>
    <cluster>events_cluster</cluster>
    <shard>01</shard>        <!-- 노드별로 다르게 설정 -->
    <replica>ch-node-1</replica>
  </macros>
</clickhouse>

-- 원시 이벤트 로컬 테이블 (각 샤드)
CREATE TABLE clickstream_events_local ON CLUSTER events_cluster
(
    event_id    UUID           DEFAULT generateUUIDv4(),
    user_id     UInt64,
    session_id  String,
    event_type  LowCardinality(String),
    page_path   String,
    revenue     Decimal(10, 2) DEFAULT 0,
    event_time  DateTime64(3)  DEFAULT now64()
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/clickstream_events',
    '{replica}'
)
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, page_path, event_time)
TTL event_time + INTERVAL 90 DAY DELETE;
 
 
-- 집계 결과 로컬 테이블 (각 샤드)
CREATE TABLE clickstream_hourly_local ON CLUSTER events_cluster
(
    event_hour    DateTime,
    event_type    LowCardinality(String),
    page_path     String,
    event_count   AggregateFunction(count),
    unique_users  AggregateFunction(uniq, UInt64),
    total_revenue AggregateFunction(sum, Decimal(10, 2))
)
ENGINE = ReplicatedAggregatingMergeTree(
    '/clickhouse/tables/{shard}/clickstream_hourly',
    '{replica}'
)
ORDER BY (event_hour, event_type, page_path);
 
 
-- MV: 로컬 원시 → 로컬 집계 (각 샤드에서 독립적으로 발화)
CREATE MATERIALIZED VIEW clickstream_mv ON CLUSTER events_cluster
TO clickstream_hourly_local
AS
SELECT
    toStartOfHour(event_time)  AS event_hour,
    event_type,
    page_path,
    countState()               AS event_count,
    uniqState(user_id)         AS unique_users,
    sumState(revenue)          AS total_revenue
FROM clickstream_events_local
GROUP BY event_hour, event_type, page_path;
 
 
-- 원시 이벤트 Distributed 테이블 (쓰기 진입점)
CREATE TABLE clickstream_events ON CLUSTER events_cluster
AS clickstream_events_local
ENGINE = Distributed(
    'events_cluster',
    currentDatabase(),
    'clickstream_events_local',
    cityHash64(user_id)
);
 
 
-- 집계 결과 Distributed 테이블 (읽기 진입점)
CREATE TABLE clickstream_hourly ON CLUSTER events_cluster
AS clickstream_hourly_local
ENGINE = Distributed(
    'events_cluster',
    currentDatabase(),
    'clickstream_hourly_local',
    rand()
);

npm install @clickhouse/client kafkajs

// clickhouse-client.ts
import { createClient, ClickHouseClient } from '@clickhouse/client';
 
export const chClient: ClickHouseClient = createClient({
  url: process.env.CLICKHOUSE_URL ?? 'http://localhost:8123',
  username: process.env.CLICKHOUSE_USER ?? 'default',
  password: process.env.CLICKHOUSE_PASSWORD ?? '',
  database: 'analytics',
  clickhouse_settings: {
    async_insert: 1,
    // 0: 서버가 메모리에 수신하면 즉시 응답 — 처리량 최대화
    // 서버 크래시 시 디스크에 기록되지 않은 배치가 유실될 수 있음
    // 데이터 유실이 허용되지 않는 경우 1로 설정
    wait_for_async_insert: 0,
  },
});

// event-ingestion.ts
import { chClient } from './clickhouse-client';
 
export interface ClickstreamEvent {
  user_id: number;
  session_id: string;
  event_type: 'page_view' | 'add_to_cart' | 'purchase';
  page_path: string;
  revenue?: number;
  event_time?: string;
}
 
// 배치 삽입 — 권장: 1,000~100,000건 단위
export async function insertEvents(events: ClickstreamEvent[]): Promise<void> {
  await chClient.insert({
    table: 'clickstream_events',
    values: events.map(e => ({
      ...e,
      revenue: e.revenue ?? 0,
      event_time: e.event_time ?? new Date().toISOString(),
    })),
    format: 'JSONEachRow',
  });
}
 
// 타입 안전 집계 쿼리 — 명명된 파라미터 사용
export async function getHourlyStats(pagePathPrefix: string, hours: number) {
  const result = await chClient.query({
    query: `
      SELECT
        event_hour,
        event_type,
        page_path,
        countMerge(event_count)   AS views,
        uniqMerge(unique_users)   AS unique_users,
        sumMerge(total_revenue)   AS revenue
      FROM clickstream_hourly
      WHERE event_hour >= now() - INTERVAL {hours:UInt32} HOUR
        AND page_path LIKE {path_prefix:String}
      GROUP BY event_hour, event_type, page_path
      ORDER BY event_hour DESC
    `,
    query_params: {
      hours,
      path_prefix: `${pagePathPrefix}%`,
    },
    format: 'JSONEachRow',
  });
 
  return result.json<{
    event_hour: string;
    event_type: string;
    page_path: string;
    views: string;
    unique_users: string;
    revenue: string;
  }>();
}

// kafka-consumer.ts
import { Kafka } from 'kafkajs';
import { insertEvents, ClickstreamEvent } from './event-ingestion';
 
const kafka = new Kafka({ brokers: ['kafka:9092'] });
const consumer = kafka.consumer({ groupId: 'clickhouse-ingester' });
 
const BATCH_SIZE = 5000;
const FLUSH_INTERVAL_MS = 1000;
 
let buffer: ClickstreamEvent[] = [];
 
async function flushBuffer(): Promise<void> {
  if (buffer.length === 0) return;
  const batch = buffer.splice(0, buffer.length);
  try {
    await insertEvents(batch);
  } catch (err) {
    console.error('ClickHouse insert failed, re-queuing batch:', err);
    // 실패한 배치를 버퍼 앞쪽으로 되돌려 다음 플러시에서 재시도
    buffer.unshift(...batch);
  }
}
 
// setInterval의 async 콜백에서 발생하는 rejection은 자동으로 전파되지 않으므로 명시적 처리
setInterval(() => {
  flushBuffer().catch(err => console.error('Flush interval error:', err));
}, FLUSH_INTERVAL_MS);
 
await consumer.subscribe({ topic: 'clickstream-events' });
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value!.toString()) as ClickstreamEvent;
    buffer.push(event);
    if (buffer.length >= BATCH_SIZE) {
      await flushBuffer();
    }
  },
});