DuckDB 분석 엔진 실전 가이드: 로컬 OLAP부터 데이터 레이크 통합까지

행 지향 (SQLite/PostgreSQL):
┌──────┬─────────┬─────────┬──────────┐
│ id   │ name    │ country │ revenue  │  -- 행 1의 모든 컬럼
├──────┼─────────┼─────────┼──────────┤
│ id   │ name    │ country │ revenue  │  -- 행 2의 모든 컬럼
└──────┴─────────┴─────────┴──────────┘
  ↑ SELECT country, SUM(revenue) 시에도
    id, name 컬럼 데이터를 불필요하게 읽음

컬럼 지향 (DuckDB):
┌────────────────────────┐
│ id:      1, 2, 3, ...  │  -- 컬럼 블록 1 (스킵)
├────────────────────────┤
│ name:    A, B, C, ...  │  -- 컬럼 블록 2 (스킵)
├────────────────────────┤
│ country: KR, US, ...   │  -- 컬럼 블록 3 (스캔!)
├────────────────────────┤
│ revenue: 100, 200, ... │  -- 컬럼 블록 4 (스캔!)
└────────────────────────┘
  ↑ 필요한 컬럼만 디스크에서 읽음
    CPU 캐시 효율 극대화

┌─────────────────────────────────────────────────────┐
│             DuckDB 벡터화 실행 엔진                  │
├─────────────────────────────────────────────────────┤
│                                                     │
│  ┌─────────────┐                                    │
│  │  SQL Parser  │  SQL 텍스트 → AST                 │
│  └──────┬──────┘                                    │
│         ▼                                           │
│  ┌─────────────┐                                    │
│  │  Binder     │  테이블/컬럼 이름 해석              │
│  └──────┬──────┘                                    │
│         ▼                                           │
│  ┌─────────────────┐                                │
│  │ Query Optimizer │  조인 순서, 필터 푸시다운        │
│  │  (Cost-based)   │  서브쿼리 플래트닝              │
│  └──────┬──────────┘                                │
│         ▼                                           │
│  ┌─────────────────────────────────────┐            │
│  │    Physical Plan (Pipeline-based)   │            │
│  │  ┌──────┐  ┌──────┐  ┌──────────┐  │            │
│  │  │Scan  │→ │Filter│→ │HashAggr  │  │            │
│  │  │(Vec) │  │(Vec) │  │  (Vec)   │  │            │
│  │  └──────┘  └──────┘  └──────────┘  │            │
│  │     ↑ 벡터(1024행) 단위 데이터 전달  │            │
│  └─────────────────────────────────────┘            │
│         ▼                                           │
│  ┌─────────────────────────────────────┐            │
│  │  Morsel-Driven Parallelism          │            │
│  │  ┌────────┐ ┌────────┐ ┌────────┐  │            │
│  │  │Thread 1│ │Thread 2│ │Thread N│  │            │
│  │  │Morsel A│ │Morsel B│ │Morsel C│  │            │
│  │  └────────┘ └────────┘ └────────┘  │            │
│  │  각 스레드가 독립적 데이터 청크 처리  │            │
│  └─────────────────────────────────────┘            │
│         ▼                                           │
│  ┌─────────────┐                                    │
│  │   Result    │  벡터 → DataFrame / Arrow Table    │
│  └─────────────┘                                    │
└─────────────────────────────────────────────────────┘

# macOS (Homebrew)
brew install duckdb

# Linux (apt)
sudo apt-get install duckdb
# 또는 직접 바이너리 다운로드
wget https://github.com/duckdb/duckdb/releases/latest/download/duckdb_cli-linux-amd64.zip
unzip duckdb_cli-linux-amd64.zip
chmod +x duckdb
sudo mv duckdb /usr/local/bin/

# 버전 확인
duckdb --version

# pip
pip install duckdb

# conda
conda install -c conda-forge python-duckdb

# 특정 버전 설치 (예: 1.2.x)
pip install duckdb==1.2.1

-- DuckDB CLI 실행
-- 인메모리 모드 (데이터는 세션 종료 시 삭제)
duckdb

-- 파일 기반 영구 저장 모드
duckdb my_analytics.db

-- 기본 쿼리 테스트
SELECT version();
-- v1.2.1

-- 시스템 정보 확인
SELECT * FROM duckdb_settings()
WHERE name IN ('threads', 'memory_limit', 'temp_directory');

-- CSV 파일 직접 쿼리
SELECT
    product_category,
    COUNT(*) AS order_count,
    ROUND(AVG(amount), 2) AS avg_amount,
    SUM(amount) AS total_revenue
FROM read_csv('sales_2025.csv', auto_detect=true)
WHERE order_date >= '2025-01-01'
GROUP BY product_category
ORDER BY total_revenue DESC
LIMIT 10;

-- 여러 CSV 파일을 글로브 패턴으로 한번에 쿼리
SELECT
    filename,
    COUNT(*) AS row_count
FROM read_csv('logs/access_*.csv', auto_detect=true, filename=true)
GROUP BY filename
ORDER BY row_count DESC;

-- JSON 파일 쿼리
SELECT
    json_extract_string(data, '$.user.name') AS user_name,
    json_extract(data, '$.events') AS events
FROM read_json('user_activity.json', auto_detect=true);

-- Parquet 파일 쿼리 (가장 빠른 포맷)
SELECT
    region,
    YEAR(event_timestamp) AS event_year,
    COUNT(*) AS event_count
FROM read_parquet('events/*.parquet')
GROUP BY region, event_year
ORDER BY event_count DESC;

-- 매출 데이터에서 이동 평균과 순위 계산
WITH daily_sales AS (
    SELECT
        sale_date,
        product_id,
        SUM(amount) AS daily_amount
    FROM read_parquet('sales_data.parquet')
    GROUP BY sale_date, product_id
)
SELECT
    sale_date,
    product_id,
    daily_amount,
    -- 7일 이동 평균
    ROUND(AVG(daily_amount) OVER (
        PARTITION BY product_id
        ORDER BY sale_date
        ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
    ), 2) AS moving_avg_7d,
    -- 제품별 매출 순위 (일별)
    RANK() OVER (
        PARTITION BY sale_date
        ORDER BY daily_amount DESC
    ) AS daily_rank,
    -- 전일 대비 증감률
    ROUND(
        (daily_amount - LAG(daily_amount) OVER (
            PARTITION BY product_id ORDER BY sale_date
        )) * 100.0 / NULLIF(LAG(daily_amount) OVER (
            PARTITION BY product_id ORDER BY sale_date
        ), 0), 2
    ) AS pct_change
FROM daily_sales
ORDER BY sale_date DESC, daily_rank
LIMIT 20;

-- 월별 카테고리별 매출 피벗 테이블
PIVOT (
    SELECT
        MONTHNAME(order_date) AS month,
        category,
        SUM(revenue) AS total
    FROM read_parquet('orders.parquet')
    WHERE YEAR(order_date) = 2025
    GROUP BY month, category
)
ON category
USING SUM(total)
GROUP BY month
ORDER BY month;

import duckdb
import pandas as pd

# Pandas DataFrame 생성
sales_df = pd.DataFrame({
    'product': ['A', 'B', 'C', 'A', 'B', 'C'] * 1000,
    'region': ['Seoul', 'Busan', 'Seoul', 'Busan', 'Seoul', 'Busan'] * 1000,
    'amount': [100, 200, 150, 300, 250, 180] * 1000,
    'date': pd.date_range('2025-01-01', periods=6000, freq='h')
})

# DataFrame을 바로 SQL로 쿼리 (변수명이 테이블명)
result = duckdb.sql("""
    SELECT
        product,
        region,
        COUNT(*) AS order_count,
        ROUND(AVG(amount), 2) AS avg_amount,
        SUM(amount) AS total_revenue,
        MIN(date) AS first_order,
        MAX(date) AS last_order
    FROM sales_df
    GROUP BY product, region
    ORDER BY total_revenue DESC
""").df()  # 결과를 다시 Pandas DataFrame으로 반환

print(result)

# 대용량 Parquet 파일과 DataFrame 조인
products_df = pd.DataFrame({
    'product': ['A', 'B', 'C'],
    'category': ['Electronics', 'Clothing', 'Food'],
    'margin_pct': [0.15, 0.40, 0.25]
})

enriched = duckdb.sql("""
    SELECT
        s.product,
        p.category,
        SUM(s.amount) AS revenue,
        ROUND(SUM(s.amount) * p.margin_pct, 2) AS estimated_profit
    FROM sales_df s
    JOIN products_df p ON s.product = p.product
    GROUP BY s.product, p.category, p.margin_pct
    ORDER BY estimated_profit DESC
""").df()

print(enriched)

import duckdb

# 영구 데이터베이스 연결
con = duckdb.connect('analytics.db')

# 테이블 생성 및 Parquet에서 데이터 로드
con.execute("""
    CREATE TABLE IF NOT EXISTS events AS
    SELECT * FROM read_parquet('raw_events/*.parquet');
""")

# Relation API (메서드 체이닝)
result = (
    con.table('events')
    .filter("event_type = 'purchase'")
    .aggregate("product_id, COUNT(*) AS cnt, SUM(amount) AS total")
    .order("total DESC")
    .limit(10)
    .df()
)

# 결과를 Parquet로 내보내기
con.execute("""
    COPY (
        SELECT product_id, COUNT(*) AS cnt, SUM(amount) AS total
        FROM events
        WHERE event_type = 'purchase'
        GROUP BY product_id
        ORDER BY total DESC
    ) TO 'top_products.parquet' (FORMAT PARQUET, COMPRESSION ZSTD);
""")

con.close()

import duckdb
import pyarrow as pa
import pyarrow.parquet as pq

# Arrow Table을 직접 쿼리
arrow_table = pq.read_table('large_dataset.parquet')

result = duckdb.sql("""
    SELECT
        category,
        COUNT(*) AS cnt,
        APPROX_QUANTILE(value, 0.95) AS p95
    FROM arrow_table
    GROUP BY category
""").arrow()  # 결과를 Arrow Table로 반환 (제로카피)

# Arrow의 제로카피 전달로 메모리 효율 극대화
print(f"Result rows: {result.num_rows}")

-- httpfs 확장 설치 및 로드
INSTALL httpfs;
LOAD httpfs;

-- AWS S3 인증 설정 (Secrets 방식 - 권장)
CREATE SECRET my_s3_secret (
    TYPE S3,
    KEY_ID 'AKIAIOSFODNN7EXAMPLE',
    SECRET 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY',
    REGION 'ap-northeast-2'
);

-- S3의 Parquet 파일 직접 쿼리
SELECT
    event_date,
    user_segment,
    COUNT(DISTINCT user_id) AS dau,
    COUNT(*) AS total_events,
    ROUND(AVG(session_duration_sec), 1) AS avg_session_sec
FROM read_parquet('s3://my-data-lake/events/year=2025/month=12/*.parquet')
GROUP BY event_date, user_segment
ORDER BY event_date DESC, dau DESC;

-- 하이브 파티셔닝된 데이터 쿼리 (파티션 프루닝 자동 적용)
SELECT *
FROM read_parquet(
    's3://my-data-lake/events/**/*.parquet',
    hive_partitioning=true
)
WHERE year = 2025 AND month = 12
  AND event_type = 'purchase';

-- 쿼리 결과를 S3에 Parquet로 다시 저장
COPY (
    SELECT
        date_trunc('hour', event_timestamp) AS hour,
        COUNT(*) AS event_count,
        COUNT(DISTINCT user_id) AS unique_users
    FROM read_parquet('s3://my-data-lake/raw_events/**/*.parquet',
                      hive_partitioning=true)
    WHERE year = 2025
    GROUP BY hour
)
TO 's3://my-data-lake/aggregated/hourly_summary.parquet'
(FORMAT PARQUET, COMPRESSION ZSTD);

-- Delta Lake 확장 설치
INSTALL delta;
LOAD delta;

-- Delta Lake 테이블 읽기
SELECT *
FROM delta_scan('s3://my-data-lake/delta_tables/user_events/')
WHERE event_date >= '2025-12-01'
LIMIT 100;

-- GCS 인증 설정
CREATE SECRET gcs_secret (
    TYPE GCS,
    KEY_ID 'my-gcs-key-id',
    SECRET 'my-gcs-secret'
);

-- GCS Parquet 파일 쿼리
SELECT COUNT(*), SUM(revenue)
FROM read_parquet('gs://my-bucket/analytics/sales_*.parquet');

-- 설치된 확장 확인
SELECT * FROM duckdb_extensions() WHERE installed = true;

-- 핵심 확장 설치
INSTALL httpfs;    -- S3/HTTP 원격 파일 접근
INSTALL json;      -- JSON 파싱 (자동 로드)
INSTALL parquet;   -- Parquet 지원 (자동 로드)
INSTALL spatial;   -- 지리공간 데이터 처리 (ST_* 함수)
INSTALL icu;       -- 국제화 정렬/비교
INSTALL fts;       -- Full-Text Search (전문 검색)
INSTALL delta;     -- Delta Lake 테이블 읽기
INSTALL excel;     -- Excel 파일 읽기/쓰기
INSTALL sqlite;    -- SQLite 데이터베이스 직접 읽기
INSTALL postgres;  -- PostgreSQL 직접 쿼리 (FDW)
INSTALL mysql;     -- MySQL 직접 쿼리

-- 확장 로드
LOAD httpfs;
LOAD spatial;

INSTALL spatial;
LOAD spatial;

-- 서울 지역 매장 반경 1km 이내 주문 분석
SELECT
    store_name,
    COUNT(*) AS nearby_orders,
    SUM(order_amount) AS nearby_revenue
FROM orders o
JOIN stores s ON o.store_id = s.id
WHERE ST_DWithin(
    ST_Point(o.longitude, o.latitude),
    ST_Point(s.longitude, s.latitude),
    0.01  -- 약 1km (WGS84 근사)
)
GROUP BY store_name
ORDER BY nearby_revenue DESC;

INSTALL postgres;
LOAD postgres;

-- PostgreSQL에 직접 연결하여 쿼리
ATTACH 'dbname=mydb user=analyst host=pg-server.internal' AS pg (TYPE POSTGRES);

-- PostgreSQL 테이블과 로컬 Parquet 파일 조인
SELECT
    p.customer_id,
    p.customer_name,
    l.total_purchases,
    l.last_purchase_date
FROM pg.public.customers p
JOIN (
    SELECT
        customer_id,
        COUNT(*) AS total_purchases,
        MAX(purchase_date) AS last_purchase_date
    FROM read_parquet('s3://data-lake/purchases/**/*.parquet')
    GROUP BY customer_id
) l ON p.customer_id = l.customer_id
WHERE l.total_purchases > 100
ORDER BY l.total_purchases DESC;

import duckdb

# MotherDuck 토큰으로 연결
con = duckdb.connect('md:my_database?motherduck_token=eyJhb...')

# 로컬 데이터와 클라우드 데이터를 하이브리드 쿼리
con.execute("""
    SELECT
        cloud_table.user_id,
        local_file.purchase_amount
    FROM my_database.users AS cloud_table
    JOIN read_parquet('local_data/purchases.parquet') AS local_file
        ON cloud_table.user_id = local_file.user_id
    WHERE cloud_table.signup_date >= '2025-01-01'
""")

result = con.fetchdf()
print(result)

-- 현재 설정 확인
SELECT name, value, description
FROM duckdb_settings()
WHERE name IN ('memory_limit', 'threads', 'temp_directory');

-- 메모리 한도 설정 (기본: 시스템 RAM의 약 80%)
SET memory_limit = '16GB';

-- 스레드 수 조정 (기본: CPU 코어 수)
-- 다른 작업과 병행 시 줄여서 사용
SET threads = 4;

-- 디스크 스필 디렉토리 설정 (메모리 초과 시 사용)
SET temp_directory = '/fast-ssd/duckdb_temp';

-- 1. EXPLAIN으로 쿼리 플랜 확인
EXPLAIN
SELECT region, SUM(amount) AS total
FROM read_parquet('large_sales.parquet')
WHERE sale_date >= '2025-01-01'
GROUP BY region;

-- 2. EXPLAIN ANALYZE로 실제 실행 프로파일링
EXPLAIN ANALYZE
SELECT region, SUM(amount) AS total
FROM read_parquet('large_sales.parquet')
WHERE sale_date >= '2025-01-01'
GROUP BY region;

-- 3. Parquet 필터 푸시다운 활용
--    WHERE 조건이 Parquet 파일의 Row Group 통계에 반영되어
--    불필요한 데이터를 읽지 않음
SELECT COUNT(*)
FROM read_parquet('partitioned_data/**/*.parquet',
                  hive_partitioning=true)
WHERE year = 2025 AND month = 12;
-- → year != 2025인 파일은 아예 열지 않음 (파티션 프루닝)

-- 4. 적절한 데이터 타입 사용
-- VARCHAR 대신 ENUM으로 카디널리티 낮은 컬럼 최적화
CREATE TYPE status_enum AS ENUM ('active', 'inactive', 'pending');

-- 5. 커넥션 재사용
-- 빈번한 connect/disconnect는 캐시 손실을 초래
-- 하나의 연결을 유지하며 쿼리 실행

-- 최적화된 Parquet 파일 생성
-- ZSTD 압축 + 적절한 Row Group 크기
COPY (
    SELECT * FROM large_table
    ORDER BY partition_key, timestamp  -- 정렬하여 압축률 극대화
) TO 'optimized_output.parquet'
(FORMAT PARQUET,
 COMPRESSION ZSTD,
 ROW_GROUP_SIZE 122880);  -- 약 120K 행/그룹

-- 파티셔닝된 Parquet 내보내기
COPY (SELECT * FROM events)
TO 'output_dir' (FORMAT PARQUET,
                  PARTITION_BY (year, month),
                  COMPRESSION ZSTD);

-- 증상: "Out of Memory Error: could not allocate block of size ..."

-- 해결 1: 메모리 한도 늘리기
SET memory_limit = '32GB';

-- 해결 2: 디스크 스필 활성화
SET temp_directory = '/tmp/duckdb_spill';

-- 해결 3: 쿼리를 청크 단위로 분할
-- 전체 테이블 집계 대신 파티션별로 처리
CREATE TABLE monthly_agg AS
SELECT * FROM (
    SELECT month, SUM(amount) AS total
    FROM read_parquet('data/year=2025/month=01/*.parquet')
    GROUP BY month
    UNION ALL
    SELECT month, SUM(amount) AS total
    FROM read_parquet('data/year=2025/month=02/*.parquet')
    GROUP BY month
    -- ... 반복
);

-- 증상: "HTTP Error: Unable to connect to URL"

-- 해결 1: 인증 정보 확인
SELECT * FROM duckdb_secrets();

-- 해결 2: 리전 명시
CREATE OR REPLACE SECRET (
    TYPE S3,
    REGION 'ap-northeast-2',
    KEY_ID 'AKIA...',
    SECRET '...'
);

-- 해결 3: 엔드포인트 커스텀 (MinIO, R2 등)
CREATE SECRET (
    TYPE S3,
    KEY_ID 'minioadmin',
    SECRET 'minioadmin',
    ENDPOINT 'localhost:9000',
    USE_SSL false,
    URL_STYLE 'path'
);

-- 증상: "Could not set lock on file ... database is locked"

-- 원인: 다른 프로세스가 동일 DB 파일을 열고 있음
-- DuckDB는 단일 Writer만 허용

-- 해결 1: access_mode를 read_only로 설정 (읽기만 필요한 경우)
-- Python에서:
-- con = duckdb.connect('my.db', read_only=True)

-- 해결 2: 인메모리 모드에서 Parquet 직접 쿼리
-- DB 파일 잠금 자체를 회피
-- con = duckdb.connect()  -- 인메모리
-- con.sql("SELECT * FROM read_parquet('data.parquet')")

import duckdb
from datetime import datetime

def daily_aggregation_job():
    """DuckDB를 ETL의 T(Transform) 단계로 활용"""
    con = duckdb.connect()

    today = datetime.now().strftime('%Y-%m-%d')

    # S3에서 원본 데이터 읽기 → 변환 → S3에 결과 저장
    con.execute(f"""
        INSTALL httpfs; LOAD httpfs;
        CREATE SECRET (TYPE S3, REGION 'ap-northeast-2');

        COPY (
            SELECT
                date_trunc('hour', event_ts) AS hour,
                event_type,
                COUNT(*) AS cnt,
                COUNT(DISTINCT user_id) AS unique_users,
                APPROX_COUNT_DISTINCT(session_id) AS approx_sessions,
                PERCENTILE_CONT(0.50) WITHIN GROUP
                    (ORDER BY duration_ms) AS p50_duration,
                PERCENTILE_CONT(0.99) WITHIN GROUP
                    (ORDER BY duration_ms) AS p99_duration
            FROM read_parquet(
                's3://raw-data/events/dt={today}/**/*.parquet'
            )
            GROUP BY hour, event_type
        )
        TO 's3://processed-data/hourly_agg/dt={today}/result.parquet'
        (FORMAT PARQUET, COMPRESSION ZSTD);
    """)

    con.close()
    print(f"Daily aggregation completed for {today}")

import duckdb
import sys

def validate_data_quality(parquet_path: str) -> bool:
    """CI/CD 파이프라인에서 데이터 품질을 자동 검증"""
    con = duckdb.connect()
    checks_passed = True

    # 검증 1: 행 수 최소 기준
    row_count = con.execute(f"""
        SELECT COUNT(*) FROM read_parquet('{parquet_path}')
    """).fetchone()[0]

    if row_count < 1000:
        print(f"FAIL: Row count {row_count} below minimum 1000")
        checks_passed = False

    # 검증 2: NULL 비율 체크
    null_check = con.execute(f"""
        SELECT
            ROUND(COUNT(*) FILTER (WHERE user_id IS NULL)
                  * 100.0 / COUNT(*), 2) AS null_pct
        FROM read_parquet('{parquet_path}')
    """).fetchone()[0]

    if null_check > 5.0:
        print(f"FAIL: user_id NULL ratio {null_check}% exceeds 5%")
        checks_passed = False

    # 검증 3: 값 범위 체크
    range_check = con.execute(f"""
        SELECT
            MIN(amount) AS min_amt,
            MAX(amount) AS max_amt
        FROM read_parquet('{parquet_path}')
    """).fetchone()

    if range_check[0] < 0:
        print(f"FAIL: Negative amount detected: {range_check[0]}")
        checks_passed = False

    con.close()
    return checks_passed

if __name__ == '__main__':
    path = sys.argv[1]
    if not validate_data_quality(path):
        sys.exit(1)
    print("All data quality checks passed")

from fastapi import FastAPI
import duckdb

app = FastAPI()

# 애플리케이션 시작 시 한번만 연결 (커넥션 재사용)
analytics_con = duckdb.connect('analytics_cache.db', read_only=True)

@app.get("/api/dashboard/summary")
async def dashboard_summary(date_from: str, date_to: str):
    result = analytics_con.execute("""
        SELECT
            product_category,
            COUNT(*) AS orders,
            ROUND(SUM(revenue), 2) AS total_revenue,
            ROUND(AVG(revenue), 2) AS avg_order_value
        FROM sales
        WHERE order_date BETWEEN ? AND ?
        GROUP BY product_category
        ORDER BY total_revenue DESC
    """, [date_from, date_to]).df()

    return result.to_dict(orient='records')