DuckDB 인메모리 분석 엔진 실전 가이드: OLAP 워크로드를 로컬에서 처리하기

-- 이 쿼리는 name과 amount 컬럼만 읽는다
-- 행 기반 DB는 모든 컬럼을 읽어야 하지만, DuckDB는 2개 컬럼만 접근
SELECT name, SUM(amount)
FROM sales
GROUP BY name;

-- 병렬 처리 설정 확인 및 조정
SELECT current_setting('threads');

-- 스레드 수 명시적 설정
SET threads TO 8;

-- 워커당 메모리 설정
SET memory_limit = '4GB';

-- 임시 디렉토리 설정 (스필 대상)
SET temp_directory = '/tmp/duckdb_spill';

-- 메모리 제한 설정 (이 값을 초과하면 디스크로 스필)
SET memory_limit = '2GB';

-- 아웃오브코어 처리로 대규모 정렬 수행
SELECT *
FROM large_table
ORDER BY timestamp_col;

# 설치
# pip install duckdb

import duckdb

# 인메모리 데이터베이스 생성
con = duckdb.connect()

# 기본 쿼리 실행
result = con.execute("""
    SELECT
        range AS id,
        'user_' || range AS name,
        random() * 1000 AS score
    FROM range(1000000)
""").fetchdf()

print(f"행 수: {len(result)}")
print(result.head())

# 집계 쿼리
con.execute("""
    CREATE TABLE sales AS
    SELECT
        range AS id,
        CASE WHEN random() < 0.3 THEN 'electronics'
             WHEN random() < 0.6 THEN 'clothing'
             ELSE 'food' END AS category,
        (random() * 500)::INTEGER AS amount,
        DATE '2025-01-01' + INTERVAL (range % 365) DAY AS sale_date
    FROM range(10000000)
""")

# 카테고리별 월간 매출 집계
result = con.execute("""
    SELECT
        category,
        DATE_TRUNC('month', sale_date) AS month,
        COUNT(*) AS cnt,
        SUM(amount) AS total_amount,
        AVG(amount)::INTEGER AS avg_amount
    FROM sales
    GROUP BY category, DATE_TRUNC('month', sale_date)
    ORDER BY month, total_amount DESC
""").fetchdf()

print(result)

# 설치 (macOS)
brew install duckdb

# 인메모리 모드로 시작
duckdb

# 파일 기반 데이터베이스로 시작
duckdb my_analytics.duckdb

-- 테이블 생성 및 데이터 삽입
CREATE TABLE metrics (
    timestamp TIMESTAMP,
    server_id VARCHAR,
    cpu_usage DOUBLE,
    memory_usage DOUBLE
);

INSERT INTO metrics
SELECT
    TIMESTAMP '2026-01-01' + INTERVAL (range * 60) SECOND,
    'server-' || (range % 10)::VARCHAR,
    50 + random() * 50,
    30 + random() * 60
FROM range(100000);

-- 서버별 CPU 사용률 통계
SELECT
    server_id,
    MIN(cpu_usage)::DECIMAL(5,2) AS min_cpu,
    AVG(cpu_usage)::DECIMAL(5,2) AS avg_cpu,
    MAX(cpu_usage)::DECIMAL(5,2) AS max_cpu,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY cpu_usage)::DECIMAL(5,2) AS p95_cpu
FROM metrics
GROUP BY server_id
ORDER BY avg_cpu DESC;

-- 시간대별 이상 탐지 (CPU 95% 이상)
SELECT
    DATE_TRUNC('hour', timestamp) AS hour,
    COUNT(*) AS anomaly_count,
    LIST(DISTINCT server_id) AS affected_servers
FROM metrics
WHERE cpu_usage > 95
GROUP BY DATE_TRUNC('hour', timestamp)
HAVING COUNT(*) > 5
ORDER BY anomaly_count DESC;

// npm install duckdb

const duckdb = require('duckdb')

// 인메모리 데이터베이스 생성
const db = new duckdb.Database(':memory:')
const con = db.connect()

// Promise 래퍼 함수
function query(sql) {
  return new Promise((resolve, reject) => {
    con.all(sql, (err, rows) => {
      if (err) reject(err)
      else resolve(rows)
    })
  })
}

async function main() {
  // Parquet 파일 직접 쿼리
  const results = await query(`
    SELECT
      category,
      COUNT(*) as count,
      SUM(amount) as total
    FROM read_parquet('sales_data.parquet')
    GROUP BY category
    ORDER BY total DESC
    LIMIT 10
  `)

  console.log('Top categories:', results)

  // CSV에서 데이터 로드 후 분석
  await query(`
    CREATE TABLE logs AS
    SELECT * FROM read_csv_auto('access_logs.csv')
  `)

  const hourlyStats = await query(`
    SELECT
      DATE_TRUNC('hour', timestamp) AS hour,
      COUNT(*) AS request_count,
      AVG(response_time_ms) AS avg_response_ms
    FROM logs
    GROUP BY DATE_TRUNC('hour', timestamp)
    ORDER BY hour
  `)

  console.log('Hourly stats:', hourlyStats)
}

main().catch(console.error)

import duckdb

con = duckdb.connect()

# 단일 Parquet 파일 쿼리
result = con.execute("""
    SELECT
        customer_region,
        COUNT(*) AS order_count,
        SUM(total_amount) AS revenue,
        AVG(total_amount)::DECIMAL(10,2) AS avg_order_value
    FROM read_parquet('orders_2025.parquet')
    WHERE order_date >= '2025-06-01'
    GROUP BY customer_region
    ORDER BY revenue DESC
""").fetchdf()

# 와일드카드로 여러 Parquet 파일 한번에 쿼리
result = con.execute("""
    SELECT
        filename,
        DATE_TRUNC('month', event_time) AS month,
        COUNT(*) AS event_count
    FROM read_parquet('events/year=2025/month=*/data_*.parquet',
                       filename=true,
                       hive_partitioning=true)
    GROUP BY filename, DATE_TRUNC('month', event_time)
""").fetchdf()

# Parquet 메타데이터만 조회 (파일을 읽지 않음)
metadata = con.execute("""
    SELECT * FROM parquet_metadata('orders_2025.parquet')
""").fetchdf()

print(f"행 그룹 수: {len(metadata)}")
print(f"컬럼: {metadata['path_in_schema'].unique()}")

# Parquet 스키마 조회
schema = con.execute("""
    SELECT * FROM parquet_schema('orders_2025.parquet')
""").fetchdf()
print(schema)

# CSV 자동 탐지 쿼리
result = con.execute("""
    SELECT *
    FROM read_csv_auto('sales_data.csv')
    LIMIT 10
""").fetchdf()

# 명시적 스키마 지정 (대규모 파일에서 안정적)
result = con.execute("""
    SELECT
        product_name,
        SUM(quantity) AS total_qty,
        SUM(price * quantity) AS total_revenue
    FROM read_csv('sales_data.csv',
        columns = {
            'order_id': 'INTEGER',
            'product_name': 'VARCHAR',
            'quantity': 'INTEGER',
            'price': 'DECIMAL(10,2)',
            'order_date': 'DATE'
        },
        dateformat = '%Y-%m-%d',
        header = true,
        delim = ','
    )
    GROUP BY product_name
    ORDER BY total_revenue DESC
    LIMIT 20
""").fetchdf()

# 여러 CSV 파일을 글로브 패턴으로 한번에 읽기
result = con.execute("""
    SELECT COUNT(*) AS total_rows
    FROM read_csv_auto('logs/access_log_2025_*.csv')
""").fetchone()
print(f"전체 로그 행 수: {result[0]:,}")

# JSON 파일 쿼리
result = con.execute("""
    SELECT
        json_extract_string(data, '$.user.name') AS user_name,
        json_extract_string(data, '$.event_type') AS event_type,
        json_extract(data, '$.metadata.duration')::INTEGER AS duration_ms
    FROM read_json_auto('events.json')
    WHERE json_extract_string(data, '$.event_type') = 'page_view'
""").fetchdf()

# NDJSON (Newline Delimited JSON) 파일 쿼리
result = con.execute("""
    SELECT
        timestamp,
        level,
        message,
        service
    FROM read_json_auto('application.ndjson', format='newline_delimited')
    WHERE level = 'ERROR'
    ORDER BY timestamp DESC
    LIMIT 100
""").fetchdf()

# Parquet 주문 데이터 + CSV 고객 데이터 + JSON 설정 데이터 조인
result = con.execute("""
    SELECT
        c.customer_name,
        c.tier,
        COUNT(o.order_id) AS order_count,
        SUM(o.amount) AS total_spent,
        AVG(o.amount)::DECIMAL(10,2) AS avg_order
    FROM read_parquet('orders/*.parquet') o
    JOIN read_csv_auto('customers.csv') c
        ON o.customer_id = c.id
    WHERE o.order_date >= '2025-01-01'
    GROUP BY c.customer_name, c.tier
    HAVING COUNT(o.order_id) >= 5
    ORDER BY total_spent DESC
    LIMIT 20
""").fetchdf()

import duckdb
import pandas as pd
import numpy as np

# pandas DataFrame 생성
df_orders = pd.DataFrame({
    'order_id': range(1, 1000001),
    'customer_id': np.random.randint(1, 10001, 1000000),
    'amount': np.random.uniform(10, 500, 1000000).round(2),
    'category': np.random.choice(['electronics', 'clothing', 'food', 'books'], 1000000),
    'order_date': pd.date_range('2025-01-01', periods=1000000, freq='30s')
})

df_customers = pd.DataFrame({
    'id': range(1, 10001),
    'name': [f'Customer_{i}' for i in range(1, 10001)],
    'tier': np.random.choice(['gold', 'silver', 'bronze'], 10000)
})

# DataFrame을 직접 SQL로 쿼리 (변수명이 테이블명이 됨)
result = duckdb.sql("""
    SELECT
        c.tier,
        o.category,
        COUNT(*) AS order_count,
        SUM(o.amount)::DECIMAL(12,2) AS total_revenue,
        AVG(o.amount)::DECIMAL(10,2) AS avg_order_value
    FROM df_orders o
    JOIN df_customers c ON o.customer_id = c.id
    GROUP BY c.tier, o.category
    ORDER BY c.tier, total_revenue DESC
""").fetchdf()

print(result)

import duckdb
import polars as pl

# polars DataFrame 생성
df_events = pl.DataFrame({
    'event_id': range(1, 5000001),
    'user_id': [i % 100000 for i in range(1, 5000001)],
    'event_type': ['click', 'view', 'purchase', 'search'] * 1250000,
    'duration_ms': [abs(int(x)) for x in (100 + 50 * i % 7 for i in range(5000000))],
    'timestamp': pl.date_range(
        pl.datetime(2025, 1, 1),
        pl.datetime(2025, 12, 31),
        eager=True
    ).sample(5000000, with_replacement=True).sort()
})

# polars DataFrame을 DuckDB로 쿼리
result = duckdb.sql("""
    SELECT
        event_type,
        DATE_TRUNC('month', timestamp) AS month,
        COUNT(*) AS event_count,
        AVG(duration_ms)::INTEGER AS avg_duration,
        PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY duration_ms)::INTEGER AS p95_duration
    FROM df_events
    GROUP BY event_type, DATE_TRUNC('month', timestamp)
    ORDER BY month, event_count DESC
""").fetchdf()

# 결과를 다시 polars로 변환
result_pl = pl.from_pandas(result)
print(result_pl)

import duckdb
import pandas as pd
import numpy as np
import time

# 1000만 행 테스트 데이터 생성
n_rows = 10_000_000
df = pd.DataFrame({
    'group_a': np.random.choice(['A', 'B', 'C', 'D', 'E'], n_rows),
    'group_b': np.random.randint(1, 101, n_rows),
    'value': np.random.uniform(0, 1000, n_rows)
})

# pandas로 그룹별 집계
start = time.time()
pandas_result = df.groupby(['group_a', 'group_b']).agg(
    count=('value', 'count'),
    sum_val=('value', 'sum'),
    mean_val=('value', 'mean'),
    std_val=('value', 'std')
).reset_index().sort_values('sum_val', ascending=False).head(20)
pandas_time = time.time() - start

# DuckDB로 동일한 집계
start = time.time()
duckdb_result = duckdb.sql("""
    SELECT
        group_a,
        group_b,
        COUNT(*) AS count,
        SUM(value)::DECIMAL(15,2) AS sum_val,
        AVG(value)::DECIMAL(10,2) AS mean_val,
        STDDEV(value)::DECIMAL(10,2) AS std_val
    FROM df
    GROUP BY group_a, group_b
    ORDER BY sum_val DESC
    LIMIT 20
""").fetchdf()
duckdb_time = time.time() - start

print(f"pandas: {pandas_time:.3f}s")
print(f"DuckDB: {duckdb_time:.3f}s")
print(f"속도 향상: {pandas_time / duckdb_time:.1f}x")
# 일반적으로 DuckDB가 3~10배 빠름

import duckdb

con = duckdb.connect()

# httpfs 익스텐션 설치 및 로드
con.execute("INSTALL httpfs")
con.execute("LOAD httpfs")

# AWS 자격증명 설정
con.execute("""
    SET s3_region = 'ap-northeast-2';
    SET s3_access_key_id = 'YOUR_ACCESS_KEY';
    SET s3_secret_access_key = 'YOUR_SECRET_KEY';
""")

# S3 Parquet 파일 직접 쿼리
result = con.execute("""
    SELECT
        DATE_TRUNC('day', event_time) AS day,
        event_type,
        COUNT(*) AS event_count
    FROM read_parquet('s3://my-data-lake/events/year=2025/month=12/*.parquet',
                       hive_partitioning=true)
    GROUP BY DATE_TRUNC('day', event_time), event_type
    ORDER BY day, event_count DESC
""").fetchdf()

print(result)

con = duckdb.connect()
con.execute("INSTALL httpfs")
con.execute("LOAD httpfs")

# HTTP URL에서 CSV 직접 쿼리
result = con.execute("""
    SELECT
        Country,
        SUM(Confirmed) AS total_confirmed,
        SUM(Deaths) AS total_deaths,
        (SUM(Deaths)::FLOAT / NULLIF(SUM(Confirmed), 0) * 100)::DECIMAL(5,2) AS fatality_rate
    FROM read_csv_auto(
        'https://raw.githubusercontent.com/datasets/covid-19/main/data/countries-aggregated.csv'
    )
    WHERE Date >= '2023-01-01'
    GROUP BY Country
    ORDER BY total_confirmed DESC
    LIMIT 20
""").fetchdf()

print(result)

import duckdb

# MotherDuck 연결 (인증 토큰 필요)
con = duckdb.connect('md:my_database')

# 로컬 데이터와 클라우드 데이터 조인
result = con.execute("""
    SELECT
        local_table.id,
        cloud_table.aggregated_value
    FROM read_parquet('local_data.parquet') AS local_table
    JOIN my_database.cloud_table
        ON local_table.id = cloud_table.id
""").fetchdf()

-- 1. EXPLAIN ANALYZE로 실행 계획 확인
EXPLAIN ANALYZE
SELECT customer_id, SUM(amount)
FROM orders
WHERE order_date >= '2025-01-01'
GROUP BY customer_id
HAVING SUM(amount) > 10000;

-- 2. 프로파일링 활성화
PRAGMA enable_profiling;
PRAGMA profiling_output = '/tmp/duckdb_profile.json';
PRAGMA profiling_mode = 'detailed';

-- 쿼리 실행 후 프로파일 확인
SELECT customer_id, SUM(amount)
FROM orders
GROUP BY customer_id;

-- 3. 통계 정보 확인
CALL pragma_storage_info('orders');

-- 4. 인덱스 활용 (DuckDB는 ART 인덱스 지원)
-- 주의: DuckDB는 OLAP 최적화 DB이므로 인덱스가 항상 유리한 것은 아님
CREATE INDEX idx_orders_date ON orders(order_date);

-- 5. 파티셔닝된 Parquet으로 내보내기 (쿼리 성능 향상)
COPY (
    SELECT * FROM orders
) TO 'orders_partitioned' (
    FORMAT PARQUET,
    PARTITION_BY (order_year, order_month),
    OVERWRITE_OR_IGNORE true
);

-- 현재 메모리 사용량 확인
SELECT * FROM duckdb_memory();

-- 메모리 제한 설정
SET memory_limit = '8GB';

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

-- 프로그레스 바 활성화 (장시간 쿼리 모니터링)
SET enable_progress_bar = true;
SET enable_progress_bar_print = true;

-- 불필요한 컬럼 제거로 메모리 절약
-- Bad: SELECT * 는 모든 컬럼을 메모리에 로드
-- Good: 필요한 컬럼만 명시
SELECT order_id, amount, order_date
FROM read_parquet('large_orders.parquet');

import duckdb

con = duckdb.connect()
con.execute("SET memory_limit = '4GB'")
con.execute("SET temp_directory = '/tmp/duckdb_spill'")

# 청크 단위 처리로 메모리 효율적 분석
con.execute("""
    CREATE TABLE daily_summary AS
    SELECT
        DATE_TRUNC('day', event_time) AS day,
        event_type,
        COUNT(*) AS cnt,
        SUM(value) AS total_value
    FROM read_parquet('s3://data-lake/events/year=2025/**/*.parquet',
                       hive_partitioning=true)
    GROUP BY DATE_TRUNC('day', event_time), event_type
""")

# 결과를 Parquet로 내보내기
con.execute("""
    COPY daily_summary
    TO 'daily_summary.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)
""")

# 처리 완료 후 메모리 해제
con.execute("DROP TABLE daily_summary")

import duckdb

con = duckdb.connect()
con.execute("INSTALL postgres_scanner")
con.execute("LOAD postgres_scanner")

# PostgreSQL에 직접 연결하여 쿼리
con.execute("""
    CALL postgres_attach(
        'host=localhost port=5432 dbname=mydb user=analyst password=secret'
    )
""")

# PostgreSQL 테이블을 DuckDB SQL로 분석
result = con.execute("""
    SELECT
        DATE_TRUNC('month', created_at) AS month,
        status,
        COUNT(*) AS order_count,
        SUM(total_amount)::DECIMAL(15,2) AS revenue
    FROM postgres_scan('public', 'orders')
    WHERE created_at >= '2025-01-01'
    GROUP BY DATE_TRUNC('month', created_at), status
    ORDER BY month
""").fetchdf()

# 로컬 Parquet와 PostgreSQL 데이터 조인
result = con.execute("""
    SELECT
        p.product_name,
        o.order_count,
        o.total_revenue
    FROM read_parquet('product_catalog.parquet') p
    JOIN (
        SELECT product_id, COUNT(*) AS order_count, SUM(amount) AS total_revenue
        FROM postgres_scan('public', 'order_items')
        GROUP BY product_id
    ) o ON p.id = o.product_id
    ORDER BY o.total_revenue DESC
    LIMIT 20
""").fetchdf()

print(result)

import duckdb

# 읽기 전용 모드로 여러 프로세스 접근
con = duckdb.connect('analytics.duckdb', read_only=True)

# 쓰기가 필요한 경우 단일 프로세스만 접근
con_write = duckdb.connect('analytics.duckdb', read_only=False)

import duckdb
import os

# 시스템 메모리의 70%를 DuckDB에 할당 (나머지는 OS와 다른 프로세스용)
total_memory_gb = os.sysconf('SC_PAGE_SIZE') * os.sysconf('SC_PHYS_PAGES') / (1024**3)
duckdb_memory = int(total_memory_gb * 0.7)

con = duckdb.connect()
con.execute(f"SET memory_limit = '{duckdb_memory}GB'")
con.execute("SET temp_directory = '/tmp/duckdb_spill'")

# 메모리 사용량 모니터링
def check_memory():
    result = con.execute("""
        SELECT
            tag,
            memory_usage_bytes / (1024*1024) AS memory_mb
        FROM duckdb_memory()
        ORDER BY memory_usage_bytes DESC
    """).fetchdf()
    print(result)
    return result

check_memory()

import duckdb
import shutil
from datetime import datetime

# 파일 기반 데이터베이스 사용 (영속성 확보)
con = duckdb.connect('production_analytics.duckdb')

# WAL(Write-Ahead Log) 모드 확인
# DuckDB는 자체 WAL 구현을 사용
con.execute("""
    CREATE TABLE IF NOT EXISTS etl_log (
        run_id INTEGER,
        table_name VARCHAR,
        rows_processed BIGINT,
        started_at TIMESTAMP,
        completed_at TIMESTAMP,
        status VARCHAR
    )
""")

# 백업 (DuckDB 파일을 직접 복사)
def backup_database(db_path, backup_dir):
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_path = f"{backup_dir}/analytics_backup_{timestamp}.duckdb"

    # 쓰기 커넥션을 닫은 후 복사
    shutil.copy2(db_path, backup_path)
    print(f"백업 완료: {backup_path}")
    return backup_path

# EXPORT DATABASE로 SQL 덤프 생성
con.execute("""
    EXPORT DATABASE '/tmp/analytics_export' (FORMAT PARQUET)
""")

import duckdb

con = duckdb.connect()

# 문제 상황: 10억 행 데이터에 대한 고카디널리티 GROUP BY
# 이 쿼리는 메모리를 과도하게 사용할 수 있다
try:
    result = con.execute("""
        SELECT user_id, COUNT(DISTINCT session_id), AVG(duration)
        FROM read_parquet('huge_events_*.parquet')
        GROUP BY user_id
    """).fetchdf()
except duckdb.OutOfMemoryException:
    print("메모리 초과 발생!")

# 해결 방법 1: 아웃오브코어 처리 활성화
con.execute("SET memory_limit = '4GB'")
con.execute("SET temp_directory = '/ssd/duckdb_temp'")  # SSD 추천

# 해결 방법 2: 단계적 처리 (partition pruning 활용)
for month in range(1, 13):
    result = con.execute(f"""
        SELECT user_id, COUNT(DISTINCT session_id), AVG(duration)
        FROM read_parquet('events/month={month:02d}/*.parquet',
                           hive_partitioning=true)
        GROUP BY user_id
    """).fetchdf()
    result.to_parquet(f'partial_results/month_{month:02d}.parquet')

# 부분 결과를 합산
final = con.execute("""
    SELECT user_id,
           SUM(count_distinct_session) AS total_sessions,
           AVG(avg_duration) AS overall_avg_duration
    FROM read_parquet('partial_results/month_*.parquet')
    GROUP BY user_id
""").fetchdf()

# 해결 방법 3: 근사 집계 사용 (정확도 vs 메모리 트레이드오프)
result = con.execute("""
    SELECT
        user_segment,
        APPROX_COUNT_DISTINCT(user_id) AS approx_users,
        APPROX_QUANTILE(amount, 0.5) AS median_amount,
        APPROX_QUANTILE(amount, 0.95) AS p95_amount
    FROM read_parquet('huge_events_*.parquet')
    GROUP BY user_segment
""").fetchdf()

import duckdb
import time
import random

# 문제 상황: 여러 ETL 잡이 동일 DB에 동시 쓰기 시도
# 해결: 쓰기는 단일 프로세스로 직렬화

# 패턴 1: 파일 잠금 + 재시도
def safe_write(db_path, query, max_retries=5):
    for attempt in range(max_retries):
        try:
            con = duckdb.connect(db_path)
            con.execute(query)
            con.close()
            return True
        except duckdb.IOException as e:
            if "lock" in str(e).lower():
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"잠금 충돌, {wait_time:.1f}초 후 재시도 ({attempt + 1}/{max_retries})")
                time.sleep(wait_time)
            else:
                raise
    raise Exception("최대 재시도 횟수 초과")

# 패턴 2: 각 프로세스가 개별 파일에 쓰고 나중에 병합
def write_to_individual_file(worker_id, data_path):
    output_path = f'worker_output/worker_{worker_id}.parquet'
    con = duckdb.connect()  # 인메모리로 처리
    con.execute(f"""
        COPY (
            SELECT * FROM read_parquet('{data_path}')
            WHERE some_condition
        ) TO '{output_path}' (FORMAT PARQUET)
    """)
    con.close()

# 병합 단계
def merge_results():
    con = duckdb.connect('final_analytics.duckdb')
    con.execute("""
        CREATE OR REPLACE TABLE results AS
        SELECT * FROM read_parquet('worker_output/worker_*.parquet')
    """)
    con.close()

import duckdb

con = duckdb.connect()

# 문제 상황: 스키마가 다른 Parquet 파일들을 함께 읽기
# 시간이 지나면서 컬럼이 추가되거나 타입이 변경된 경우

# 해결 방법 1: union_by_name 옵션 사용
result = con.execute("""
    SELECT *
    FROM read_parquet(
        'events/year=*/month=*/*.parquet',
        hive_partitioning = true,
        union_by_name = true  -- 컬럼명 기준으로 자동 합치기
    )
    LIMIT 10
""").fetchdf()

# 해결 방법 2: 스키마 확인 후 안전한 쿼리 작성
schema_info = con.execute("""
    SELECT file_name, name, type
    FROM parquet_schema('events/year=2025/month=01/*.parquet')
""").fetchdf()
print("스키마 확인:", schema_info)

# 해결 방법 3: COALESCE로 누락 컬럼 처리
result = con.execute("""
    SELECT
        event_id,
        event_type,
        COALESCE(new_column, 'default_value') AS new_column,
        timestamp
    FROM read_parquet(
        'events/**/*.parquet',
        union_by_name = true
    )
""").fetchdf()

import duckdb
from datetime import datetime

def daily_etl_pipeline(date_str):
    """일별 ETL 파이프라인: S3 원본 -> 변환 -> Parquet 저장"""
    con = duckdb.connect()
    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';
    """)

    # Step 1: 원본 데이터 읽기 + 클린징
    con.execute(f"""
        CREATE TEMP TABLE raw_events AS
        SELECT
            event_id,
            LOWER(TRIM(event_type)) AS event_type,
            user_id,
            COALESCE(amount, 0) AS amount,
            TRY_CAST(timestamp AS TIMESTAMP) AS event_time,
            properties
        FROM read_parquet(
            's3://raw-data/events/date={date_str}/*.parquet'
        )
        WHERE TRY_CAST(timestamp AS TIMESTAMP) IS NOT NULL
          AND user_id IS NOT NULL
    """)

    # Step 2: 사용자 세그먼트 매핑
    con.execute("""
        CREATE TEMP TABLE enriched_events AS
        SELECT
            e.*,
            u.segment,
            u.registration_date,
            DATEDIFF('day', u.registration_date, e.event_time) AS days_since_registration
        FROM raw_events e
        LEFT JOIN read_parquet('s3://dim-data/users/latest/*.parquet') u
            ON e.user_id = u.user_id
    """)

    # Step 3: 집계 테이블 생성
    con.execute(f"""
        COPY (
            SELECT
                DATE_TRUNC('hour', event_time) AS hour,
                segment,
                event_type,
                COUNT(*) AS event_count,
                COUNT(DISTINCT user_id) AS unique_users,
                SUM(amount) AS total_amount,
                AVG(amount)::DECIMAL(10,2) AS avg_amount
            FROM enriched_events
            GROUP BY
                DATE_TRUNC('hour', event_time),
                segment,
                event_type
        ) TO 's3://processed-data/hourly_summary/date={date_str}/data.parquet'
        (FORMAT PARQUET, COMPRESSION ZSTD)
    """)

    row_count = con.execute("SELECT COUNT(*) FROM enriched_events").fetchone()[0]
    print(f"[{date_str}] ETL 완료: {row_count:,}행 처리")
    con.close()
    return row_count

# 실행
daily_etl_pipeline('2025-12-15')

import duckdb

def validate_data_quality(parquet_path):
    """데이터 품질 검증 쿼리 모음"""
    con = duckdb.connect()

    # 기본 통계
    stats = con.execute(f"""
        SELECT
            COUNT(*) AS total_rows,
            COUNT(DISTINCT user_id) AS unique_users,
            MIN(event_time) AS earliest_event,
            MAX(event_time) AS latest_event,
            COUNT(*) FILTER (WHERE user_id IS NULL) AS null_user_count,
            COUNT(*) FILTER (WHERE amount < 0) AS negative_amount_count,
            COUNT(*) FILTER (WHERE event_type NOT IN
                ('click', 'view', 'purchase', 'search')) AS unknown_event_count
        FROM read_parquet('{parquet_path}')
    """).fetchdf()

    print("=== 기본 통계 ===")
    print(stats.to_string(index=False))

    # 중복 검사
    duplicates = con.execute(f"""
        SELECT
            event_id,
            COUNT(*) AS dup_count
        FROM read_parquet('{parquet_path}')
        GROUP BY event_id
        HAVING COUNT(*) > 1
        ORDER BY dup_count DESC
        LIMIT 10
    """).fetchdf()

    if len(duplicates) > 0:
        print(f"\n[경고] 중복 이벤트 발견: {len(duplicates)}건")
        print(duplicates)
    else:
        print("\n[통과] 중복 이벤트 없음")

    # 시간 연속성 검사 (빈 시간대 탐지)
    gaps = con.execute(f"""
        WITH hourly AS (
            SELECT DATE_TRUNC('hour', event_time) AS hour, COUNT(*) AS cnt
            FROM read_parquet('{parquet_path}')
            GROUP BY DATE_TRUNC('hour', event_time)
        ),
        expected_hours AS (
            SELECT UNNEST(generate_series(
                (SELECT MIN(hour) FROM hourly),
                (SELECT MAX(hour) FROM hourly),
                INTERVAL 1 HOUR
            )) AS hour
        )
        SELECT e.hour AS missing_hour
        FROM expected_hours e
        LEFT JOIN hourly h ON e.hour = h.hour
        WHERE h.hour IS NULL
        ORDER BY e.hour
    """).fetchdf()

    if len(gaps) > 0:
        print(f"\n[경고] 데이터 누락 시간대: {len(gaps)}건")
        print(gaps.head(10))
    else:
        print("\n[통과] 시간 연속성 정상")

    con.close()

validate_data_quality('events/year=2025/month=12/*.parquet')

import duckdb

def create_dashboard_views(db_path):
    """BI 대시보드용 사전 집계 뷰 생성"""
    con = duckdb.connect(db_path)

    # 일별 KPI 뷰
    con.execute("""
        CREATE OR REPLACE VIEW v_daily_kpi AS
        SELECT
            DATE_TRUNC('day', order_time) AS date,
            COUNT(DISTINCT customer_id) AS dau,
            COUNT(*) AS total_orders,
            SUM(amount)::DECIMAL(15,2) AS gmv,
            AVG(amount)::DECIMAL(10,2) AS aov,
            COUNT(*) FILTER (WHERE is_first_order) AS new_customer_orders,
            COUNT(*) FILTER (WHERE is_refunded) AS refund_count,
            (COUNT(*) FILTER (WHERE is_refunded)::FLOAT /
             NULLIF(COUNT(*), 0) * 100)::DECIMAL(5,2) AS refund_rate
        FROM orders
        GROUP BY DATE_TRUNC('day', order_time)
        ORDER BY date
    """)

    # 코호트 분석 뷰
    con.execute("""
        CREATE OR REPLACE VIEW v_cohort_retention AS
        WITH first_purchase AS (
            SELECT
                customer_id,
                DATE_TRUNC('month', MIN(order_time)) AS cohort_month
            FROM orders
            GROUP BY customer_id
        ),
        monthly_activity AS (
            SELECT
                customer_id,
                DATE_TRUNC('month', order_time) AS activity_month
            FROM orders
            GROUP BY customer_id, DATE_TRUNC('month', order_time)
        )
        SELECT
            fp.cohort_month,
            DATEDIFF('month', fp.cohort_month, ma.activity_month) AS months_since_first,
            COUNT(DISTINCT ma.customer_id) AS active_users,
            COUNT(DISTINCT ma.customer_id)::FLOAT /
                NULLIF(COUNT(DISTINCT fp.customer_id) FILTER (
                    WHERE DATEDIFF('month', fp.cohort_month, ma.activity_month) = 0
                ), 0) AS retention_rate
        FROM first_purchase fp
        JOIN monthly_activity ma ON fp.customer_id = ma.customer_id
        GROUP BY fp.cohort_month,
                 DATEDIFF('month', fp.cohort_month, ma.activity_month)
        ORDER BY fp.cohort_month, months_since_first
    """)

    # 퍼널 분석 뷰
    con.execute("""
        CREATE OR REPLACE VIEW v_conversion_funnel AS
        WITH step1 AS (
            SELECT DISTINCT user_id, DATE_TRUNC('day', event_time) AS day
            FROM events WHERE event_type = 'page_view'
        ),
        step2 AS (
            SELECT DISTINCT user_id, DATE_TRUNC('day', event_time) AS day
            FROM events WHERE event_type = 'add_to_cart'
        ),
        step3 AS (
            SELECT DISTINCT user_id, DATE_TRUNC('day', event_time) AS day
            FROM events WHERE event_type = 'checkout'
        ),
        step4 AS (
            SELECT DISTINCT user_id, DATE_TRUNC('day', event_time) AS day
            FROM events WHERE event_type = 'purchase'
        )
        SELECT
            s1.day,
            COUNT(DISTINCT s1.user_id) AS page_view_users,
            COUNT(DISTINCT s2.user_id) AS add_to_cart_users,
            COUNT(DISTINCT s3.user_id) AS checkout_users,
            COUNT(DISTINCT s4.user_id) AS purchase_users,
            (COUNT(DISTINCT s4.user_id)::FLOAT /
             NULLIF(COUNT(DISTINCT s1.user_id), 0) * 100)::DECIMAL(5,2) AS overall_conversion_rate
        FROM step1 s1
        LEFT JOIN step2 s2 ON s1.user_id = s2.user_id AND s1.day = s2.day
        LEFT JOIN step3 s3 ON s2.user_id = s3.user_id AND s2.day = s3.day
        LEFT JOIN step4 s4 ON s3.user_id = s4.user_id AND s3.day = s4.day
        GROUP BY s1.day
        ORDER BY s1.day
    """)

    print("대시보드 뷰 생성 완료")
    con.close()

create_dashboard_views('analytics.duckdb')

import duckdb

con = duckdb.connect('analytics.duckdb')

# 1. 메모리 사용량 모니터링
memory_info = con.execute("""
    SELECT
        tag,
        (memory_usage_bytes / 1024 / 1024)::INTEGER AS usage_mb
    FROM duckdb_memory()
    WHERE memory_usage_bytes > 0
    ORDER BY memory_usage_bytes DESC
""").fetchdf()
print("메모리 사용량:", memory_info.to_string(index=False))

# 2. 테이블 크기 확인
table_sizes = con.execute("""
    SELECT
        table_name,
        estimated_size / 1024 / 1024 AS estimated_size_mb,
        column_count,
        index_count
    FROM duckdb_tables()
    ORDER BY estimated_size DESC
""").fetchdf()
print("테이블 크기:", table_sizes.to_string(index=False))

# 3. 실행 중인 쿼리 확인 (DuckDB 1.1+)
# DuckDB는 단일 프로세스이므로, 장시간 쿼리가 다른 작업을 차단할 수 있다
# enable_progress_bar로 진행 상황 확인 가능

# 4. 데이터베이스 파일 크기 확인
import os
db_size = os.path.getsize('analytics.duckdb') / (1024 * 1024)
print(f"DB 파일 크기: {db_size:.1f} MB")

con.close()