ClickHouse에 데이터 로드: 배치, async_insert, 모니터링

ClickHouse에 데이터 로드: 한 행씩 삽입을 멈추고 처리 속도를 500배 향상시킨 방법

재난 이야기: 초당 1000개 쿼리와 프로덕션 장애

저는 실시간 베팅 분석 프로젝트를 진행 중이었습니다. 매일 5000만 건의 이벤트가 들어왔습니다. INSERT INTO bets VALUES (...)로 한 번에 하나씩 처리해도 괜찮다고 생각했습니다. ClickHouse는 초당 500개의 삽입을 처리했지만, 우리는 2000개가 필요했습니다. 서버가 멈췄습니다: 디스크 큐가 폭주했고, 백그라운드 병합이 수천 개의 작은 파트를 생성했습니다. 프로덕션이 다운되었습니다.

알고 보니 ClickHouse는 MySQL이 아니었습니다. 대량 삽입을 위해 설계되었으며, 개별 INSERT용이 아닙니다. 로더를 10만 개 레코드씩 배치로 다시 작성한 후, 성능 저하 없이 초당 1만 개의 삽입을 달성했습니다.

아래는 작은 CSV부터 프로덕션급 배처까지 모든 로드 방법과 제가 밤을 새게 만든 함정들입니다.

1. 단일 행이 왜 해로운가 (원하더라도)

ClickHouse는 INSERT마다 디스크에 새로운 파트를 생성합니다. 파트는 8192행의 그레인을 포함하지만, 단 1행을 삽입해도 작은 .bin 파일이 있는 새 파트가 물리적으로 생성됩니다.

결과:

• 1행씩 1만 번 INSERT = 1만 개의 파트
• SELECT 쿼리는 1만 개의 파일을 열어야 함
• 백그라운드 병합이 이를 결합하려고 시도하여 CPU에 부담
• 파트 메타데이터로 디스크 공간 낭비

제 경험상 규칙: 배치는 최소 1000행 또는 1MB의 압축 데이터여야 합니다.

2. INSERT ... VALUES — 테스트용, 프로덕션용 아님

-- 로컬 실험 전용
INSERT INTO betting.bets (user_id, created_at, amount, odds, sport, outcome)
VALUES 
    (1001, now(), 50.00, 2.1, 'football', 'win'),
    (1002, now(), 100.00, 1.8, 'basketball', 'loss'),
    (1003, now(), 75.00, 3.0, 'tennis', 'win');

프로덕션에서는 절대 이렇게 하지 마세요. VALUES에 1000행을 배치하더라도 SQL 파싱이 병목이 됩니다.

3. INSERT ... FORMAT CSV / JSONEachRow / TSV — 파일에 생명을 불어넣는 방법

ClickHouse는 여러 형식을 즉시 이해합니다.

CSV 파일 예시 bets.csv:

1001,2024-03-15 10:00:00,50.00,2.1,football,win
1002,2024-03-15 10:01:00,100.00,1.8,basketball,loss
1003,2024-03-15 10:02:00,75.00,3.0,tennis,win

로드:

INSERT INTO betting.bets FORMAT CSV

하지만 데이터를 stdin으로 파이프해야 합니다. 따라서 스크립트에서는 다음을 사용하세요:

clickhouse-client --query="INSERT INTO betting.bets FORMAT CSV" < bets.csv

JSONEachRow — 로그에 이상적:

{"user_id":1001,"created_at":"2024-03-15 10:00:00","amount":50.00,"odds":2.1,"sport":"football","outcome":"win"}
{"user_id":1002,"created_at":"2024-03-15 10:01:00","amount":100.00,"odds":1.8,"sport":"basketball","outcome":"loss"}

로드:

clickhouse-client --query="INSERT INTO betting.bets FORMAT JSONEachRow" < bets.json

TabSeparated — 가장 빠름 (파싱이 적음):

clickhouse-client --query="INSERT INTO betting.bets FORMAT TSV" < bets.tsv

4. 파일에서 clickhouse-client를 통한 로드 — 백업에 제일 좋아하는 방법

# 작은 파일
clickhouse-client --query="INSERT INTO betting.bets FORMAT CSV" < /tmp/daily_bets.csv

# 진행 상황 표시와 함께 큰 파일
clickhouse-client \
  --query="INSERT INTO betting.bets FORMAT CSV" \
  --progress \
  --max_insert_block_size=100000 \
  < /data/historical_bets_2023.csv

중요 플래그: --max_insert_block_size=100000 — 파일을 10만 행 블록으로 분할합니다. 이를 통해 메모리 오버플로 없이 테라바이트 크기의 파일을 삽입할 수 있습니다.

배운 점: FORMAT은 입력 데이터의 구조를 정의하며 출력이 아닙니다. 쿼리 자체에는 데이터가 없으며 stdin을 통해 들어옵니다.

5. POST를 사용한 HTTP API — 명령줄 접근이 없을 때

# curl을 통한 CSV
curl -X POST "http://localhost:8123/?query=INSERT+INTO+betting.bets+FORMAT+CSV" \
  --data-binary @bets.csv

# JSONEachRow
curl -X POST "http://localhost:8123/?query=INSERT+INTO+betting.bets+FORMAT+JSONEachRow" \
  --data-binary @bets.json \
  -u analyst:password

# 실시간 GZIP (대역폭 절약)
gzip -c bets.csv | curl -X POST "http://localhost:8123/?query=INSERT+INTO+betting.bets+FORMAT+CSV" \
  --data-binary @- \
  --header "Content-Encoding: gzip"

왜 POST인가, GET이 아닌가: GET 요청은 데이터를 포함하여 완전히 로깅됩니다. ?query=INSERT...를 통해 삽입하면 비밀번호와 데이터가 프록시 로그에 남을 수 있습니다.

6. INSERT SELECT — ClickHouse 내부에서 데이터 이동

대량 재로드의 가장 빠른 방법은 외부로 내보내지 않는 것입니다.

-- 1월 데이터를 아카이브 테이블로 복사
INSERT INTO betting.bets_archive
SELECT * FROM betting.bets
WHERE created_at >= '2024-01-01' AND created_at < '2024-02-01';

-- 실시간 집계
INSERT INTO betting.daily_stats (date, total_bets, total_amount)
SELECT 
    toDate(created_at) AS date,
    count() AS total_bets,
    sum(amount) AS total_amount
FROM betting.bets
WHERE created_at >= today() - 7
GROUP BY date;

속도: INSERT SELECT는 클라이언트-서버 교환을 피하며 로컬 전송입니다. 수십억 행을 몇 분 안에 이동할 수 있습니다.

7. 비동기 삽입: 플레이어로부터 초당 수천 개의 이벤트가 올 때

기존 INSERT는 동기식입니다: 클라이언트는 데이터가 디스크에 기록될 때까지 기다립니다. 초당 1만 개의 삽입에서는 부담이 됩니다.

async_insert는 메모리에 데이터를 버퍼링하고 배치로 삽입합니다.

config.xml 또는 SETTINGS를 통해 활성화:

-- 세션 수준
SET async_insert = 1;
SET wait_for_async_insert = 0;  -- 완료를 기다리지 않음
SET async_insert_max_data_size = 10000000;  -- 10 MB 버퍼
SET async_insert_busy_timeout_ms = 200;     -- 200ms마다 플러시

INSERT INTO betting.bets FORMAT JSONEachRow
{"user_id":1001,"amount":50,"odds":2.1}
{"user_id":1002,"amount":100,"odds":1.8}

작동 방식:

1. 클라이언트가 작은 삽입을 보냄
2. ClickHouse가 메모리 버퍼에 축적
3. 버퍼가 가득 차거나(10MB) 타임아웃(200ms)이 만료되면 단일 대량 삽입 발생

나를 태운 점: wait_for_async_insert=0은 클라이언트가 삽입 실패 여부를 알 수 없음을 의미합니다. 네트워크 문제로 데이터 손실이 발생할 수 있습니다. 신뢰성이 중요하면 wait_for_async_insert=1로 설정하세요. 성능은 떨어지지만 치명적이지는 않습니다.

8. 설정: max_insert_block_size와 max_batch_size

<!-- config.xml -->
<max_insert_block_size>1048576</max_insert_block_size>  <!-- 100만 행 -->
<max_block_size>65536</max_block_size>

max_insert_block_size — ClickHouse가 한 번에 처리하는 최대 블록 크기입니다. 삽입이 더 크면 블록으로 분할됩니다.

프로덕션 규칙: max_insert_block_size를 예상 배치 크기보다 2~4배 작게 설정하세요. 그러면 실수로 큰 INSERT가 발생해도 메모리가 죽지 않습니다.

# 단일 삽입에 강제 적용
clickhouse-client --query="INSERT INTO bets FORMAT CSV" \
  --max_insert_block_size=500000 < huge_file.csv

9. system.query_log를 통한 삽입 모니터링

얼마나 많은 행이 삽입되었는지 추측하지 마세요. system.query_log를 확인하세요.

-- 최근 INSERT 쿼리
SELECT 
    event_time,
    query_duration_ms,
    read_rows,
    written_rows,
    result_rows,
    memory_usage,
    query
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query LIKE '%INSERT INTO betting.bets%'
  AND event_time >= now() - INTERVAL 1 HOUR
ORDER BY event_time DESC
LIMIT 20;

내 모니터링 대시보드: 지난 1시간 동안의 평균 배치 크기, 실패한 삽입 수, 초당 행 속도를 보여줍니다.

SELECT 
    toStartOfMinute(event_time) AS minute,
    count() AS inserts,
    avg(written_rows) AS avg_batch_size,
    sum(written_rows) AS total_rows,
    avg(query_duration_ms) AS avg_latency_ms
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query LIKE '%INSERT INTO betting.bets%'
  AND event_time >= now() - INTERVAL 1 HOUR
GROUP BY minute
ORDER BY minute DESC;

10. 과거 데이터 배치 로드를 위한 Python 스크립트

실제 프로젝트에서 S3의 parquet 파일에서 3년치 베팅 내역을 로드했습니다. 다음은 20억 행을 처리한 스크립트입니다:

#!/usr/bin/env python3
from clickhouse_driver import Client
import pandas as pd
import glob
from tqdm import tqdm

# ClickHouse에 연결
client = Client(
    host='clickhouse.prod.internal',
    port=9000,
    user='loader',
    password='strong_password',
    database='betting'
)

# 대량 삽입을 위한 최적화
client.execute("SET max_insert_block_size = 1000000")
client.execute("SET min_insert_block_size_rows = 500000")
client.execute("SET min_insert_block_size_bytes = 100000000")  # 100 MB

def insert_batch(df, table='bets'):
    """Native 프로토콜을 통해 DataFrame을 배치로 삽입"""
    # clickhouse-driver가 자동으로 타입 변환
    client.execute(f"INSERT INTO {table} FORMAT TabSeparated", 
                   df.to_csv(sep='\t', header=False, index=False),
                   types_check=True)

def stream_parquet_files(pattern='/data/bets_*.parquet'):
    files = glob.glob(pattern)
    batch_size = 500000
    buffer = []
    
    for file in tqdm(files, desc="파일 처리 중"):
        # 메모리에 모두 로드하지 않도록 parquet을 청크 단위로 읽기
        for chunk in pd.read_parquet(file, chunksize=batch_size):
            # ClickHouse에 맞게 타입 변환
            chunk['created_at'] = pd.to_datetime(chunk['created_at'])
            chunk['amount'] = chunk['amount'].astype('float64')
            chunk['odds'] = chunk['odds'].astype('float64')
            
            buffer.append(chunk)
            if len(buffer) >= 5:  # 5개 청크(각 50만 행) = 250만 행
                df_batch = pd.concat(buffer, ignore_index=True)
                insert_batch(df_batch)
                buffer = []
    
    # 남은 데이터
    if buffer:
        df_batch = pd.concat(buffer, ignore_index=True)
        insert_batch(df_batch)

if __name__ == '__main__':
    stream_parquet_files()

requests를 사용한 대안 (pandas 없이):

import requests
import gzip
import io

def insert_via_http_csv_gz(filepath):
    with open(filepath, 'rb') as f:
        compressed = gzip.compress(f.read())
    
    response = requests.post(
        'http://localhost:8123/?query=INSERT+INTO+betting.bets+FORMAT+CSV',
        data=compressed,
        headers={'Content-Encoding': 'gzip'},
        auth=('loader', 'password')
    )
    
    if response.status_code != 200:
        raise Exception(f"삽입 실패: {response.text}")
    
    print(f"{filepath} 삽입 완료")

# 여러 파일을 배치로 전송하는 예시
for file in glob.glob('/data/bets_*.csv.gz'):
    insert_via_http_csv_gz(file)

일반적인 로드 오류 (그리고 해결 방법)

오류: Memory limit (total) exceeded
원인: 삽입 블록이 너무 큼.
해결 방법: max_insert_block_size를 줄이거나 파일을 여러 조각으로 분할.

오류: Too many parts
원인: 너무 많은 작은 INSERT 또는 높은 처리량에서 async_insert 비활성화.
해결 방법: async_insert 활성화, min_rows_for_wide_part 증가.

오류: Code: 252. DB::Exception: Too many simultaneous inserts
원인: 동일한 파티션에 대한 병렬 삽입.
해결 방법: 해당 테이블에 max_insert_threads = 1 설정.

오류: 파티션이 병합되지 않음
원인: ORDER BY 순서에 맞지 않게 데이터 삽입.
해결 방법: 데이터가 테이블의 ORDER BY와 동일하게 정렬되었는지 확인.

다음은?

이제 작은 CSV부터 프로덕션 스트림까지 모든 방법으로 데이터를 로드하는 방법을 알게 되었습니다. 다음 글에서는 쿼리 최적화와 고급 집계에 대해 다룹니다.

마지막 조언: 프로덕션에서 작업한다면 어떤 INSERT도 max_insert_block_size와 async_insert 없이 스크립트를 떠나지 마세요. 저는 그 레이크를 세 번 밟았습니다 — 충분합니다.

---

← 이전 글: ClickHouse의 MergeTree: 엔진이 분석을 과립으로 나누고 파트를 병합하는 방법
→ 다음 글: ClickHouse의 SELECT 쿼리: 10년간 PostgreSQL을 사용하다가 사고방식을 바꾼 방법

— Editorial Team