AI API 게이트웨이를 운영하다 보면 요청 로그, 토큰 사용량, 응답 시간, 비용 추적 데이터를 효과적으로 저장하고 조회해야 하는 문제가 발생합니다. 이 튜토리얼에서는 ClickHouse를 사용하여 AI API 호출 히스토리를 로컬에 저장하고 최적화하는 방법을 단계별로 설명드리겠습니다.
시작하기 전에: 실제 발생했던 문제
제 경험상 AI API 게이트웨이 운영 시 가장 흔히 마주치는 문제는 바로 대규모 로그 데이터 조회 시 타임아웃입니다:
# 실제 발생했던 오류 시나리오
$ curl "https://internal-api-gateway/logs/query" -d '{"start":"2024-01-01","end":"2024-01-31"}'
{"error": "ConnectionError: timeout after 30000ms", "code": "QUERY_TIMEOUT"}
월 1억 건 이상의 API 호출 로그를 PostgreSQL에서 조회하려 했을 때, 단순한 기간 검색조차 30초 이상 소요되었고 결국 타임아웃으로 실패했습니다. 이 문제를 해결하기 위해 ClickHouse 마이그레이션을 결정했고, 결과적으로 30초 걸리던 쿼리를 200ms 이하로 단축할 수 있었습니다.
왜 ClickHouse인가?
AI API 히스토리 데이터 저장소로 ClickHouse를 선택하는 핵심 이유:
- 압축 효율성: 로그 데이터 압축률 10:1 이상 (특히 반복되는 텍스트, JSON)
- 写入 속도: 초당 100만 행以上的 쓰기 처리 가능
- 집계 쿼리 성능: GROUP BY, COUNT, SUM 연산이 매우 빠름
- 열 기반 저장: 특정 컬럼 조회 시 전체 테이블 스캔 방지
- 비용 절감:Same hardware로 10배 많은 데이터 처리 가능
ClickHouse 설치 및 기본 설정
Docker Compose로 ClickHouse 서버 구성
# docker-compose.yml
version: '3.8'
services:
clickhouse:
image: clickhouse/clickhouse-server:23.8-alpine
container_name: ai-api-clickhouse
hostname: clickhouse
ports:
- "8123:8123" # HTTP 인터페이스
- "9000:9000" # TCP 인터페이스 (Native client용)
volumes:
- ./data:/var/lib/clickhouse
- ./logs:/var/log/clickhouse
- ./config.d:/etc/clickhouse-server/config.d
environment:
CLICKHOUSE_DB: aiapi_logs
CLICKHOUSE_USER: admin
CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD}
ulimits:
nofile:
soft: 262144
hard: 262144
healthcheck:
test: ["CMD", "wget", "--spider", "-q", "localhost:8123/ping"]
interval: 10s
timeout: 5s
retries: 5
clickhouse-ui:
image: clickhouse/clickhouse-server:23.8-alpine
depends_on:
- clickhouse
entrypoint: ["sleep", "infinity"]
# config.d/custom-config.xml
<?xml version="1.0"?>
<clickhouse>
<max_concurrent_queries>100</max_concurrent_queries>
<max_server_memory_usage>0</max_server_memory_usage>
<max_thread_pool_size>10000</max_thread_pool_size>
<logger>
<level>information</level>
<log>/var/log/clickhouse/clickhouse-server.log</log>
<errorlog>/var/log/clickhouse/clickhouse-server.err.log</errorlog>
<size>1000M</size>
<count>10</count>
</logger>
<query_log>
<database>system</database>
<table>query_log</table>
<partition_by>toYYYYMM(event_date)</partition_by>
<flush_interval_milliseconds>7500</flush_interval_milliseconds>
</query_log>
<metrics>
<!-- Prometheus 메트릭 활성화 -->
</metrics>
<listen_host>0.0.0.0</listen_host>
</clickhouse>
# ClickHouse 서버 시작
$ docker-compose up -d clickhouse
상태 확인
$ docker-compose ps
컨테이너 내부 접속
$ docker exec -it ai-api-clickhouse clickhouse-client --host localhost --user admin --password ${CLICKHOUSE_PASSWORD}
AI API 로그 테이블 스키마 설계
핵심 테이블: api_request_logs
-- 메인 API 요청 로그 테이블
CREATE TABLE IF NOT EXISTS aiapi_logs.api_request_logs
(
-- 기본 식별자
id UUID DEFAULT generateUUIDv4(),
request_id String DEFAULT generateUUIDv4(),
-- 시간 정보 (파티셔닝 핵심)
timestamp DateTime64(3) DEFAULT now64(3),
date Date DEFAULT toDate(timestamp),
-- API 요청 정보
provider Enum8('openai' = 1, 'anthropic' = 2, 'google' = 3, 'deepseek' = 4, 'other' = 99),
model String,
endpoint String,
method String DEFAULT 'POST',
-- 요청 페이로드 (압축 효율성)
request_tokens UInt32 DEFAULT 0,
prompt_tokens UInt32 DEFAULT 0,
completion_tokens UInt32 DEFAULT 0,
-- 응답 정보
response_time_ms UInt32,
status_code UInt16,
error_code String DEFAULT '',
error_message String DEFAULT '',
-- 비용 정보 (비용 최적화 핵심)
input_cost_usd Float64 DEFAULT 0,
output_cost_usd Float64 DEFAULT 0,
total_cost_usd Float64 DEFAULT 0,
-- 메타데이터
user_id String DEFAULT '',
api_key_id String DEFAULT '',
project_id String DEFAULT '',
-- 추가 정보 (JSONとして保存)
metadata String DEFAULT '{}'
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, timestamp, request_id)
TTL date + INTERVAL 90 DAY
SETTINGS index_granularity = 8192;
--Aggregated view for daily/monthly reports
CREATE MATERIALIZED VIEW IF NOT EXISTS aiapi_logs.daily_cost_summary
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, provider, model)
AS
SELECT
date,
provider,
model,
count() as request_count,
sum(request_tokens) as total_request_tokens,
sum(completion_tokens) as total_completion_tokens,
sum(input_cost_usd) as total_input_cost,
sum(output_cost_usd) as total_output_cost,
sum(total_cost_usd) as total_cost,
avg(response_time_ms) as avg_response_time_ms,
quantile(0.95)(response_time_ms) as p95_response_time,
countIf(status_code >= 400) as error_count
FROM aiapi_logs.api_request_logs
GROUP BY date, provider, model;
-- 실시간 모니터링용 시계열 테이블
CREATE TABLE IF NOT EXISTS aiapi_logs.realtime_metrics
(
timestamp DateTime,
provider String,
metric_name String,
metric_value Float64,
dimensions String DEFAULT '{}'
)
ENGINE = ReplacingMergeTree(timestamp)
ORDER BY (provider, metric_name, timestamp)
TTL timestamp + INTERVAL 7 DAY;
HolySheep AI API 연동 코드
이제 HolySheep AI를 통해 AI API를 호출하고 로그를 ClickHouse에 저장하는 통합 코드를 작성하겠습니다.
#!/usr/bin/env python3
"""
HolySheep AI API 로그 수집기 + ClickHouse 저장소
Author: HolySheep AI Technical Team
"""
import json
import time
import uuid
import asyncio
from datetime import datetime, date
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, asdict
from enum import Enum
import httpx
from clickhouse_driver import Client as ClickHouseClient
import os
from dotenv import load_dotenv
load_dotenv()
============================================
HolySheep AI API Configuration
============================================
class AIProvider(Enum):
OPENAI = "openai"
ANTHROPIC = "anthropic"
GOOGLE = "google"
DEEPSEEK = "deepseek"
@dataclass
class APIRequestLog:
"""API 요청 로그 데이터 클래스"""
request_id: str
timestamp: datetime
provider: str
model: str
endpoint: str
request_tokens: int
completion_tokens: int
response_time_ms: int
status_code: int
error_code: str
error_message: str
input_cost_usd: float
output_cost_usd: float
total_cost_usd: float
user_id: str
metadata: str
class HolySheepAIClient:
"""HolySheep AI API 클라이언트 + 로깅 통합"""
BASE_URL = "https://api.holysheep.ai/v1"
# 모델별 비용 (USD per 1M tokens)
MODEL_COSTS = {
# OpenAI
"gpt-4.1": {"input": 8.0, "output": 32.0},
"gpt-4.1-mini": {"input": 1.5, "output": 6.0},
"gpt-4o": {"input": 2.5, "output": 10.0},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
# Anthropic
"claude-sonnet-4.5": {"input": 15.0, "output": 75.0},
"claude-3.5-sonnet": {"input": 15.0, "output": 75.0},
"claude-3.5-haiku": {"input": 0.80, "output": 4.0},
# Google
"gemini-2.5-flash": {"input": 2.50, "output": 10.0},
"gemini-2.0-flash": {"input": 0.10, "output": 0.40},
# DeepSeek
"deepseek-v3.2": {"input": 0.42, "output": 2.08},
"deepseek-chat": {"input": 0.27, "output": 1.10},
}
def __init__(self, api_key: str, clickhouse_client: ClickHouseClient):
self.api_key = api_key
self.clickhouse = clickhouse_client
self.client = httpx.AsyncClient(
base_url=self.BASE_URL,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
timeout=60.0
)
def _calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> tuple:
"""토큰 사용량 기반 비용 계산"""
costs = self.MODEL_COSTS.get(model, {"input": 0.0, "output": 0.0})
input_cost = (prompt_tokens / 1_000_000) * costs["input"]
output_cost = (completion_tokens / 1_000_000) * costs["output"]
return input_cost, output_cost, input_cost + output_cost
def _detect_provider(self, model: str) -> str:
"""모델명 기반 Provider 감지"""
model_lower = model.lower()
if "gpt" in model_lower or "openai" in model_lower:
return "openai"
elif "claude" in model_lower or "anthropic" in model_lower:
return "anthropic"
elif "gemini" in model_lower or "google" in model_lower:
return "google"
elif "deepseek" in model_lower:
return "deepseek"
return "other"
async def chat_completion(
self,
model: str,
messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 4096,
user_id: str = "",
project_id: str = "",
metadata: Dict = None
) -> Dict[str, Any]:
"""HolySheep AI Chat Completion API 호출 + 자동 로깅"""
request_id = str(uuid.uuid4())
start_time = time.time()
log_entry = {
"request_id": request_id,
"timestamp": datetime.now(),
"provider": self._detect_provider(model),
"model": model,
"endpoint": "/chat/completions",
"request_tokens": 0,
"completion_tokens": 0,
"response_time_ms": 0,
"status_code": 200,
"error_code": "",
"error_message": "",
"input_cost_usd": 0.0,
"output_cost_usd": 0.0,
"total_cost_usd": 0.0,
"user_id": user_id,
"metadata": json.dumps(metadata or {})
}
try:
# HolySheep AI API 호출
response = await self.client.post(
"/chat/completions",
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
)
elapsed_ms = int((time.time() - start_time) * 1000)
log_entry["response_time_ms"] = elapsed_ms
log_entry["status_code"] = response.status_code
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
log_entry["request_tokens"] = usage.get("prompt_tokens", 0)
log_entry["completion_tokens"] = usage.get("completion_tokens", 0)
# 비용 계산
input_cost, output_cost, total_cost = self._calculate_cost(
model,
log_entry["request_tokens"],
log_entry["completion_tokens"]
)
log_entry["input_cost_usd"] = input_cost
log_entry["output_cost_usd"] = output_cost
log_entry["total_cost_usd"] = total_cost
return {"success": True, "data": data, "log_id": request_id}
else:
error_data = response.json() if response.content else {}
log_entry["error_code"] = error_data.get("error", {}).get("code", "UNKNOWN")
log_entry["error_message"] = error_data.get("error", {}).get("message", str(response.text))
return {"success": False, "error": log_entry["error_message"], "log_id": request_id}
except httpx.TimeoutException as e:
log_entry["status_code"] = 408
log_entry["error_code"] = "TIMEOUT"
log_entry["error_message"] = f"Request timeout after {60}s"
return {"success": False, "error": str(e), "log_id": request_id}
except httpx.HTTPStatusError as e:
log_entry["status_code"] = e.response.status_code
log_entry["error_code"] = f"HTTP_{e.response.status_code}"
log_entry["error_message"] = str(e)
return {"success": False, "error": str(e), "log_id": request_id}
except Exception as e:
log_entry["status_code"] = 500
log_entry["error_code"] = "INTERNAL_ERROR"
log_entry["error_message"] = str(e)
return {"success": False, "error": str(e), "log_id": request_id}
finally:
# 항상 ClickHouse에 로그 저장
self._save_log(log_entry)
def _save_log(self, log_entry: Dict):
"""ClickHouse에 로그 저장"""
try:
self.clickhouse.execute(
"""
INSERT INTO aiapi_logs.api_request_logs (
request_id, timestamp, provider, model, endpoint,
request_tokens, completion_tokens, response_time_ms,
status_code, error_code, error_message,
input_cost_usd, output_cost_usd, total_cost_usd,
user_id, metadata
) VALUES
""",
[{
"request_id": log_entry["request_id"],
"timestamp": log_entry["timestamp"],
"provider": log_entry["provider"],
"model": log_entry["model"],
"endpoint": log_entry["endpoint"],
"request_tokens": log_entry["request_tokens"],
"completion_tokens": log_entry["completion_tokens"],
"response_time_ms": log_entry["response_time_ms"],
"status_code": log_entry["status_code"],
"error_code": log_entry["error_code"],
"error_message": log_entry["error_message"],
"input_cost_usd": log_entry["input_cost_usd"],
"output_cost_usd": log_entry["output_cost_usd"],
"total_cost_usd": log_entry["total_cost_usd"],
"user_id": log_entry["user_id"],
"metadata": log_entry["metadata"]
}]
)
except Exception as e:
print(f"Failed to save log: {e}")
#!/usr/bin/env python3
"""
ClickHouse 쿼리 최적화 및 분석 예제
"""
from clickhouse_driver import Client
from datetime import datetime, timedelta
import json
class ClickHouseQueryOptimizer:
"""ClickHouse 쿼리 최적화 유틸리티"""
def __init__(self, host: str = "localhost", port: int = 9000,
user: str = "admin", password: str = ""):
self.client = Client(
host=host, port=port, user=user, password=password,
settings={
'max_block_size': 100000,
'max_execution_time': 300,
'use_uncompressed_cache': 1
}
)
def get_daily_cost_report(self, start_date: str, end_date: str) -> list:
"""일일 비용 리포트 조회 (Optimized Query)"""
query = """
SELECT
date,
provider,
model,
count() as total_requests,
sum(total_cost_usd) as daily_cost,
sum(request_tokens) as total_tokens,
avg(response_time_ms) as avg_latency,
quantile(0.99)(response_time_ms) as p99_latency,
countIf(status_code >= 400) as error_count
FROM aiapi_logs.api_request_logs
WHERE date BETWEEN '{start}' AND '{end}'
GROUP BY date, provider, model
ORDER BY date DESC, daily_cost DESC
""".format(start=start_date, end=end_date)
return self.client.execute(query)
def get_error_analysis(self, hours: int = 24) -> list:
"""최근 N시간 내 에러 분석"""
query = """
SELECT
error_code,
error_message,
count() as count,
provider,
model,
avg(response_time_ms) as avg_response_time
FROM aiapi_logs.api_request_logs
WHERE
timestamp >= now() - INTERVAL {hours} HOUR
AND status_code >= 400
GROUP BY error_code, error_message, provider, model
ORDER BY count DESC
LIMIT 100
""".format(hours=hours)
return self.client.execute(query)
def get_token_usage_trend(self, days: int = 30) -> list:
"""토큰 사용량 트렌드 (시간별 집계)"""
query = """
SELECT
toStartOfHour(timestamp) as hour,
provider,
sum(request_tokens) as prompt_tokens,
sum(completion_tokens) as completion_tokens,
sum(total_cost_usd) as cost
FROM aiapi_logs.api_request_logs
WHERE timestamp >= now() - INTERVAL {days} DAY
GROUP BY hour, provider
ORDER BY hour DESC
""".format(days=days)
return self.client.execute(query)
def get_model_performance_comparison(self) -> list:
"""모델별 성능 비교 (Aggregated View 활용)"""
query = """
SELECT
provider,
model,
sum(request_count) as total_requests,
avg(avg_response_time_ms) as avg_response_time,
avg(p95_response_time) as avg_p95_response_time,
sum(total_cost) as total_cost,
sum(error_count) as total_errors
FROM aiapi_logs.daily_cost_summary
WHERE date >= today() - INTERVAL 7 DAY
GROUP BY provider, model
ORDER BY total_cost DESC
"""
return self.client.execute(query)
def get_user_cost_breakdown(self, user_id: str, days: int = 30) -> dict:
"""특정 사용자의 비용 상세 분석"""
query = """
WITH (
SELECT sum(total_cost_usd)
FROM aiapi_logs.api_request_logs
WHERE user_id = '{user}' AND date >= today() - {days}
) as total_user_cost
SELECT
model,
count() as request_count,
sum(total_cost_usd) as cost,
sum(total_cost_usd) / total_user_cost * 100 as cost_percentage,
avg(response_time_ms) as avg_latency
FROM aiapi_logs.api_request_logs
WHERE user_id = '{user}' AND date >= today() - {days}
GROUP BY model
ORDER BY cost DESC
""".format(user=user_id, days=days)
rows = self.client.execute(query)
return {
"user_id": user_id,
"period_days": days,
"model_breakdown": rows
}
============================================
실제 사용 예제
============================================
if __name__ == "__main__":
# ClickHouse 연결
optimizer = ClickHouseQueryOptimizer(
host="localhost",
user="admin",
password=os.getenv("CLICKHOUSE_PASSWORD", "")
)
# 1. 최근 7일 일일 비용 리포트
print("=== Daily Cost Report (Last 7 Days) ===")
report = optimizer.get_daily_cost_report(
start_date=(datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d"),
end_date=datetime.now().strftime("%Y-%m-%d")
)
for row in report[:10]:
print(f"Date: {row[0]}, Provider: {row[1]}, Model: {row[2]}, "
f"Requests: {row[3]}, Cost: ${row[4]:.4f}")
# 2. 에러 분석
print("\n=== Error Analysis (Last 24 Hours) ===")
errors = optimizer.get_error_analysis(hours=24)
for row in errors[:5]:
print(f"Code: {row[0]}, Count: {row[2]}, Provider: {row[3]}")
# 3. 모델 성능 비교
print("\n=== Model Performance Comparison ===")
perf = optimizer.get_model_performance_comparison()
for row in perf[:5]:
print(f"Provider: {row[0]}, Model: {row[1]}, "
f"Requests: {row[2]}, Avg Latency: {row[3]:.0f}ms")
쿼리 성능 최적화 기법
1. 인덱스와 파티셔닝 전략
-- ============================================
-- ClickHouse 최적화: Skip Index 활용
-- ============================================
-- 브루트포스 인덱스: 특정 값 검색 가속
ALTER TABLE aiapi_logs.api_request_logs
ADD INDEX idx_provider provider TYPE set(100) GRANULARITY 4;
ALTER TABLE aiapi_logs.api_request_logs
ADD INDEX idx_model model TYPE bloom_filter(0.01) GRANULARITY 4;
ALTER TABLE aiapi_logs.api_request_logs
ADD INDEX idx_status status_code TYPE minmax GRANULARITY 4;
-- 인덱스 빌드 트리거
ALTER TABLE aiapi_logs.api_request_logs MATERIALIZE INDEX idx_provider;
ALTER TABLE aiapi_logs.api_request_logs MATERIALIZE INDEX idx_model;
-- ============================================
-- ClickHouse 최적화: 데이터 압축 및 저장 설정
-- ============================================
-- 테이블 재생성 시 압축 코덱 최적화
CREATE TABLE IF NOT EXISTS aiapi_logs.api_request_logs_optimized
(
id UUID DEFAULT generateUUIDv4(),
request_id String,
timestamp DateTime64(3) CODEC(ZSTD(3)),
date Date CODEC(Delta, ZSTD(3)),
provider Enum8('openai' = 1, 'anthropic' = 2, 'google' = 3, 'deepseek' = 4, 'other' = 99)
CODEC(ZSTD(2)),
model String CODEC(ZSTD(3)),
endpoint String CODEC(ZSTD(2)),
request_tokens UInt32 CODEC(Delta, ZSTD(2)),
completion_tokens UInt32 CODEC(Delta, ZSTD(2)),
response_time_ms UInt32 CODEC(Delta, ZSTD(1)),
status_code UInt16 CODEC(Delta),
error_code String CODEC(ZSTD(2)),
error_message String CODEC(ZSTD(3)),
input_cost_usd Float64 CODEC(GORILLA),
output_cost_usd Float64 CODEC(GORILLA),
total_cost_usd Float64 CODEC(GORILLA),
user_id String CODEC(ZSTD(2)),
api_key_id String CODEC(ZSTD(2)),
project_id String CODEC(ZSTD(2)),
metadata String CODEC(ZSTD(4))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, provider, timestamp)
TTL date + INTERVAL 90 DAY
SETTINGS
index_granularity = 8192,
min_bytes_for_wide_part = 10485760,
max_bytes_to_merge_at_min_space_in_pool = 5368709120;
-- 데이터 마이그레이션
INSERT INTO aiapi_logs.api_request_logs_optimized
SELECT * FROM aiapi_logs.api_request_logs;
RENAME TABLE aiapi_logs.api_request_logs TO aiapi_logs.api_request_logs_old,
aiapi_logs.api_request_logs_optimized TO aiapi_logs.api_request_logs;
2. 샘플링 쿼리 (대규모 데이터 분석)
-- ============================================
-- 샘플링을 통한 빠른 전체 통계
-- ============================================
-- 1% 샘플링으로 빠른 근사값 획득
SELECT
provider,
model,
count() * 100 as estimated_total_requests,
sum(total_cost_usd) * 100 as estimated_total_cost,
avg(response_time_ms) as avg_latency
FROM aiapi_logs.api_request_logs
SAMPLE 0.01
WHERE date >= today() - 30
GROUP BY provider, model;
-- 정확한 카운트가 필요할 때
SELECT
provider,
uniqExact(request_id) as exact_request_count,
sum(total_cost_usd) as exact_total_cost
FROM aiapi_logs.api_request_logs
WHERE date >= today() - 30
GROUP BY provider;
모니터링 대시보드 설정
# Prometheus 메트릭 내보내기 설정
/etc/prometheus/prometheus.yml
global:
scrape_interval: 15s
evaluation_interval: 15s
scrape_configs:
- job_name: 'clickhouse'
static_configs:
- targets: ['localhost:9363']
- job_name: 'ai-api-gateway'
static_configs:
- targets: ['localhost:9090']
-- ClickHouse 시스템 테이블 쿼리 (모니터링용)
-- 현재 실행 중인 쿼리 확인
SELECT
query_id,
query,
memory_usage,
elapsed,
read_rows,
read_bytes
FROM system.processes
WHERE query LIKE '%api_request_logs%'
ORDER BY elapsed DESC
LIMIT 10;
-- 파티션 정보 확인
SELECT
partition,
partition_id,
name as part_name,
active,
rows,
bytes_on_disk,
data_compressed_bytes,
data_uncompressed_bytes,
marks
FROM system.parts
WHERE table = 'api_request_logs' AND database = 'aiapi_logs'
ORDER BY modification_time DESC
LIMIT 20;
-- 디스크 사용량 분석
SELECT
database,
table,
formatReadableSize(sum(bytes)) as total_size,
formatReadableSize(sum(data_compressed_bytes)) as compressed_size,
round(sum(data_compressed_bytes) / sum(data_uncompressed_bytes) * 100, 2) as compression_ratio
FROM system.parts
WHERE table = 'api_request_logs'
GROUP BY database, table;
데이터 보존 및 백업 전략
# ============================================
ClickHouse 백업 스크립트 (S3 연동)
============================================
#!/bin/bash
backup-clickhouse.sh
set -e
BACKUP_DATE=$(date +%Y%m%d_%H%M%S)
S3_BUCKET="s3://your-backup-bucket/clickhouse"
CLICKHOUSE_HOST="${CLICKHOUSE_HOST:-localhost}"
CLICKHOUSE_PORT="${CLICKHOUSE_PORT:-9000}"
echo "[$(date)] Starting ClickHouse backup..."
1. 테이블reeze (읽기 전용 전환)
clickhouse-client --host $CLICKHOUSE_HOST --port $CLICKHOUSE_PORT \
--query "ALTER TABLE aiapi_logs.api_request_logs FREEZE;"
2. 백업 디렉토리 생성
BACKUP_DIR="/tmp/clickhouse_backup_${BACKUP_DATE}"
mkdir -p $BACKUP_DIR
3. shadow 디렉토리에서 백업 복사
cp -r /var/lib/clickhouse/data/aiapi_logs/api_request_logs/shadow/* $BACKUP_DIR/
4. S3에 업로드
aws s3 sync $BACKUP_DIR ${S3_BUCKET}/${BACKUP_DATE}/ \
--storage-class STANDARD_IA \
--exclude "*.tmp"
5. 임시 파일 정리
rm -rf $BACKUP_DIR
6. 오래된 백업 정리 (30일 이상)
aws s3 ls $S3_BUCKET | while read backup; do
backup_date=$(echo $backup | awk '{print $2}')
# 30일 이전 백업 삭제 로직
done
echo "[$(date)] Backup completed: ${S3_BUCKET}/${BACKUP_DATE}/"
이런 팀에 적합 / 비적합
| ClickHouse 기반 Tardis 데이터 저장소가 | |
|---|---|
| ✅ 적합한 팀 | ❌ 비적합한 팀 |
|
|
가격과 ROI
| 구성 요소 | 사양 | 월간 비용估算 | 비고 |
|---|---|---|---|
| ClickHouse 서버 | 8 vCPU, 32GB RAM, 500GB SSD | $150~$300 | AWS r5.large / GCP n2-standard-8 |
| 스토리지 (90일) | 월 5억 로그 × 90일 ≈ 45TB | $450~$900 | S3 + CloudFront 고려 |
| 네트워크 | 데이터転送량 기반 | $50~$200 | 쿼리 결과 내보내기 볼륨에 따라 |
| 인건비 (DevOps) | 주 4시간 관리 | $400~$800 | 설정, 백업, 모니터링 |
| 총 월간 비용 | $1,050~$2,200 | 팀 규모 및 데이터량에 따라 차등 |