DuckDB 실전 가이드 — 서버 없는 인메모리 SQL 분석, Parquet 파이프라인, MCP 에이전트 연동

DuckDB 설치 및 기본 쿼리
pip install duckdb

import duckdb

# 인메모리 DB 연결
con = duckdb.connect()

# CSV 직접 쿼리 (파일 로드 불필요)
result = con.execute("""
  SELECT category, COUNT(*) as cnt, AVG(price) as avg_price
  FROM 'data/products.csv'
  GROUP BY category
  ORDER BY cnt DESC
""").fetchdf()

print(result)

# 영구 DB 파일 사용
con = duckdb.connect('analytics.duckdb')

Parquet 직접 쿼리 및 파티셔닝
import duckdb

con = duckdb.connect()

# S3 Parquet 직접 쿼리 (boto3 불필요)
con.execute("INSTALL httpfs; LOAD httpfs;")
con.execute("""
  SET s3_region='ap-northeast-2';
  SET s3_access_key_id='YOUR_KEY';
  SET s3_secret_access_key='YOUR_SECRET';
""")

result = con.execute("""
  SELECT year, month, SUM(revenue) as total
  FROM 's3://my-bucket/sales/year=*/month=*/*.parquet'
  WHERE year >= 2025
  GROUP BY year, month
  ORDER BY year, month
""").fetchdf()

# Hive 파티셔닝으로 내보내기
con.execute("""
  COPY (
    SELECT * FROM result_table
  ) TO 'output/' (
    FORMAT PARQUET,
    PARTITION_BY (year, month),
    OVERWRITE_OR_IGNORE 1
  )
""")

Pandas / Polars 제로카피 쿼리
import duckdb
import pandas as pd
import polars as pl

con = duckdb.connect()

# Pandas DataFrame을 SQL로 직접 쿼리
df_pd = pd.DataFrame({
    'user_id': [1, 2, 3, 4, 5],
    'event': ['click', 'purchase', 'click', 'view', 'purchase'],
    'amount': [0, 29.9, 0, 0, 49.9]
})

result = con.execute("""
  SELECT event, COUNT(*) as cnt, SUM(amount) as revenue
  FROM df_pd
  GROUP BY event
""").fetchdf()

# Polars로 출력
result_pl = con.execute("""
  SELECT * FROM df_pd WHERE amount > 0
""").pl()  # Polars DataFrame으로 반환

# 벡터화 UDF 등록
def categorize(amount: float) -> str:
    return 'paid' if amount > 0 else 'free'

con.create_function('categorize', categorize, ['DOUBLE'], 'VARCHAR')
con.execute("SELECT user_id, categorize(amount) FROM df_pd").fetchdf()

DuckDB MCP 서버 설정 (Claude Code / Claude Desktop)
# DuckDB MCP 서버 설치
npx @duckdb/duckdb-mcp --help

# Claude Desktop용 claude_desktop_config.json
{
  "mcpServers": {
    "duckdb": {
      "command": "npx",
      "args": [
        "-y",
        "@duckdb/duckdb-mcp",
        "--db-path",
        "/path/to/analytics.duckdb"
      ]
    }
  }
}

# Claude Code 프로젝트 MCP (.mcp.json)
{
  "mcpServers": {
    "duckdb": {
      "command": "npx",
      "args": ["-y", "@duckdb/duckdb-mcp", "--db-path", "./analytics.duckdb"]
    }
  }
}

DuckLake 설정 및 사용
# DuckLake 확장 설치
import duckdb
con = duckdb.connect('warehouse.duckdb')
con.execute("INSTALL ducklake; LOAD ducklake;")

# 레이크하우스 생성
con.execute("""
  ATTACH 'ducklake:data/catalog.db?data_path=data/lake/' AS lake
""")

# 테이블 생성 및 INSERT
con.execute("""
  CREATE TABLE lake.sales (
    order_id BIGINT,
    product VARCHAR,
    amount DOUBLE,
    created_at TIMESTAMP
  )
""")
con.execute("INSERT INTO lake.sales VALUES (1, 'Widget', 29.9, NOW())")

# 시간 여행 쿼리
con.execute("""
  SELECT * FROM lake.sales
  AT (VERSION => 1)
""").fetchdf()

# 스냅샷 체크포인트 (오래된 버전 정리)
con.execute("CHECKPOINT lake")

DuckDB 프로덕션 성능 설정
import duckdb

con = duckdb.connect('analytics.duckdb', config={
    'threads': 8,                  # CPU 코어 수에 맞게 조정
    'memory_limit': '4GB',         # 메모리 상한 설정
    'temp_directory': '/fast-ssd', # 스필 디렉토리는 SSD로
    'preserve_insertion_order': False  # 집계 성능 향상
})

# 쿼리 프로파일링
con.execute("PRAGMA enable_profiling='json'")
con.execute("PRAGMA profiling_output='/tmp/profile.json'")

# 느린 쿼리 분석
con.execute("EXPLAIN ANALYZE SELECT ...")

# JSON 분석 가속 (DuckDB 1.5+ VARIANT 타입)
con.execute("""
  CREATE TABLE events AS
  SELECT id, payload::JSON as payload
  FROM 'events.json'
""")