DuckDB実践 — ノートPCで完結する組み込み分析の時代

+--------------------------------------------------------------+
|  ユーザープロセス (Python, CLI, Node, JVM, ...)                |
|                                                              |
|   +------------------------------------------------------+   |
|   |  DuckDB (ライブラリとして組み込み、サーバーなし)          |   |
|   |                                                      |   |
|   |  SQLパーサー / オプティマイザ                           |   |
|   |        |                                             |   |
|   |        v                                             |   |
|   |  ベクトル化実行エンジン                                  |   |
|   |   - 一度に行1つではなくベクトル(約2048個の値)を処理       |   |
|   |   - CPUキャッシュフレンドリー、SIMD活用                  |   |
|   |   - 全コアへの自動並列化                                |   |
|   |        |                                             |   |
|   |        v                                             |   |
|   |  列指向ストレージ                                       |   |
|   |   - 単一ファイルDB (ACID, MVCC)                        |   |
|   |   - 列ごとの圧縮 (辞書、RLE、ビットパッキング)            |   |
|   |   - Parquet/CSV/JSONの直接スキャン (インポート不要)      |   |
|   +------------------------------------------------------+   |
|                                                              |
|  メモリを超えるデータはディスクへスピル (out-of-core実行)        |
+--------------------------------------------------------------+

# CLIのインストール (macOS)
brew install duckdb

# インメモリモードで即実行
duckdb

# ファイルDBとして開く (なければ作成)
duckdb analytics.duckdb

# Python: pip install duckdb の1行で完了
import duckdb

# 接続なしで直接クエリ (インメモリ)
duckdb.sql("SELECT 42 AS answer").show()

# ファイルDBの使用
con = duckdb.connect("analytics.duckdb")
con.sql("CREATE TABLE t AS SELECT * FROM range(1000000)")

-- CSVを直接クエリ (スキーマ自動推論)
SELECT count(*) FROM 'events-2026-06.csv';

-- 複数のParquetファイルをグロブパターンで一括処理
SELECT user_id, sum(amount) AS total
FROM 'data/sales/*.parquet'
GROUP BY user_id
ORDER BY total DESC
LIMIT 10;

-- JSON Linesも直接
SELECT payload.event_type, count(*)
FROM read_json_auto('logs/*.jsonl')
GROUP BY 1;

-- ファイル形式間の変換もSQL 1行
COPY (SELECT * FROM 'raw.csv') TO 'clean.parquet' (FORMAT parquet);

INSTALL httpfs;
LOAD httpfs;

-- HTTP URLを直接クエリ
SELECT * FROM 'https://example.com/open-data/stats.parquet' LIMIT 5;

-- S3バケットのParquetをノートPCから直接集計
SET s3_region = 'ap-northeast-2';
SELECT date_trunc('day', ts) AS d, count(*)
FROM 's3://my-bucket/events/year=2026/month=06/*.parquet'
GROUP BY d ORDER BY d;

-- 日次売上の7日移動平均と前日比
WITH daily AS (
  SELECT date_trunc('day', ts) AS d, sum(amount) AS revenue
  FROM 'sales/*.parquet'
  GROUP BY d
)
SELECT
  d,
  revenue,
  avg(revenue) OVER (
    ORDER BY d ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
  ) AS ma7,
  revenue - lag(revenue) OVER (ORDER BY d) AS diff_prev_day,
  rank() OVER (ORDER BY revenue DESC) AS best_day_rank
FROM daily
ORDER BY d;

-- カテゴリ別の月次売上をワイドフォーマットに
PIVOT (
  SELECT category, strftime(ts, '%Y-%m') AS month, amount
  FROM 'sales/*.parquet'
)
ON month
USING sum(amount)
GROUP BY category;

-- 逆方向のUNPIVOTもサポート
UNPIVOT monthly_wide
ON COLUMNS(* EXCLUDE (category))
INTO NAME month VALUE amount;

-- 各取引に対して「取引時点より前の最新為替レート」をマッチング
SELECT t.trade_id, t.ts, t.amount_usd, r.rate AS usdkrw_at_trade
FROM trades t
ASOF JOIN fx_rates r
  ON t.ts >= r.ts;

import duckdb
import pandas as pd

df = pd.read_parquet("events.parquet")

# pandas DataFrameをコピーなしでSQLクエリ (変数名をそのまま認識)
result = duckdb.sql("""
    SELECT user_id, count(*) AS cnt
    FROM df
    WHERE event_type = 'purchase'
    GROUP BY user_id
    HAVING cnt >= 3
""").df()  # 結果を再びpandasへ

# Polarsとも同様に往復可能
import polars as pl
pdf = duckdb.sql("SELECT * FROM 'big.parquet'").pl()  # Arrow経由、コピー最小化

-- 空間分析: GIS関数とジオメトリ型
INSTALL spatial;
LOAD spatial;
SELECT ST_Distance(
  ST_Point(126.9780, 37.5665),  -- ソウル
  ST_Point(139.6917, 35.6895)   -- 東京
) AS deg_distance;

-- 全文検索: BM25ベースのインデックス
INSTALL fts;
LOAD fts;
PRAGMA create_fts_index('docs', 'doc_id', 'body');
SELECT doc_id, fts_main_docs.match_bm25(doc_id, 'duckdb embedded') AS score
FROM docs ORDER BY score DESC LIMIT 10;

# コンテナ1つで完結する日次バッチETLの骨格
import duckdb

con = duckdb.connect()
con.sql("INSTALL httpfs; LOAD httpfs;")
con.sql("""
    COPY (
        SELECT
            cast(ts AS DATE) AS event_date,
            user_id,
            event_type,
            amount
        FROM 's3://raw-bucket/events/2026-06-11/*.jsonl'
        WHERE event_type IN ('purchase', 'refund')
    )
    TO 's3://curated-bucket/events/dt=2026-06-11/data.parquet'
    (FORMAT parquet, COMPRESSION zstd)
""")

-- ローカルCSVとMotherDuckクラウドテーブルの結合 (概念例)
ATTACH 'md:my_database';
SELECT l.campaign, m.user_segment, count(*)
FROM 'local_campaign.csv' l
JOIN my_database.users m USING (user_id)
GROUP BY 1, 2;

-- 1. SUMMARIZE: テーブルプロファイリングを一発で
-- 各列のmin、max、近似ユニーク数、null比率まで自動要約
SUMMARIZE SELECT * FROM 'events.parquet';

-- 2. EXCLUDE / REPLACE: 列が多いときの救世主
SELECT * EXCLUDE (raw_payload, internal_flag)
FROM wide_table;

SELECT * REPLACE (round(amount, 2) AS amount)
FROM sales;

-- 3. 重複排除の定石: グループごとに最新1件だけ残す
SELECT * FROM events
QUALIFY row_number() OVER (
  PARTITION BY user_id ORDER BY ts DESC
) = 1;

-- 4. 再現可能なサンプリング
SELECT * FROM big_table USING SAMPLE 1 PERCENT (bernoulli, 42);

-- 5. リスト集約と構造体: 非正規化データの取り扱い
SELECT user_id, list(distinct category) AS categories
FROM purchases GROUP BY user_id;

-- 壊れた行を捨てながら読み、どの行が捨てられたか確認
SELECT * FROM read_csv('messy.csv', ignore_errors = true);
SELECT * FROM read_csv('messy.csv', store_rejects = true);
SELECT * FROM reject_errors;  -- 拒否された行と理由のテーブル

-- 1. メモリとスレッドを明示的に管理
SET memory_limit = '8GB';
SET threads = 8;

-- 2. 実行計画を確認する習慣
EXPLAIN ANALYZE
SELECT category, sum(amount) FROM 'sales/*.parquet' GROUP BY 1;

-- 3. CSVは一度Parquetに変換しておくと以降のすべてのクエリが速くなる
COPY (SELECT * FROM 'raw/*.csv') TO 'data.parquet'
  (FORMAT parquet, COMPRESSION zstd);

-- 4. パーティションディレクトリ構造を活用したプルーニング
-- year=2026/month=06 のようなhiveパーティショニングを認識する
SELECT count(*) FROM read_parquet('events/*/*/*.parquet', hive_partitioning = true)
WHERE year = 2026 AND month = 6;