第13回 DuckDBとPolars

科目：ビッグデータ応用Ⅱ〔特徴量エンジニアリング〕 担当：小野原彩香 対象：3年生（後期）

---

学習目標

この回を終えると、以下のことができるようになります。

1. DuckDBがどのような場面で適切かを説明できる
2. Polarsの遅延評価とArrow互換性のメリットを理解する
3. DuckDBの地理空間拡張（spatial）でST_Contains/ST_Distanceなどを使える
4. DuckDB→Polarsのシームレスな連携パターンを実装できる
5. PandasよりもPolars/DuckDBが速い理由を説明できる

---

1. DuckDBの特徴と使い所

1.1 DuckDBとは

DuckDBはインプロセスOLAP（Online Analytical Processing）データベースです。

• インプロセス：サーバーを別途起動せず、Pythonプロセスの中で動作する
• OLAP：大量データの集計・分析に特化（OLTPのトランザクション処理とは対照的）
• 列指向ストレージ：集計クエリで必要な列だけ読み込むため高速

イメージとしては「SQLが使えるPandas」ですが、大規模データでは圧倒的に高速です。

1.2 DuckDBが得意なこと

用途	具体例
大規模CSV/Parquetの直接読み込み	ファイルをデータベースにインポートせず、そのままSQLで集計
複雑なSQLでの集計	GROUP BY, WINDOW関数, サブクエリを組み合わせた分析
地理空間クエリ	ST_Distance, ST_Contains等の空間フィルタリング
複数ファイルの結合	複数のParquetファイルをJOINで統合
メモリ効率	RAMに収まらないデータも、一時ファイルを使って処理できる

1.3 インストールと基本操作

# インストール
pip install duckdb

# バージョン確認
python -c "import duckdb; print(duckdb.__version__)"

import duckdb

# インメモリ接続（セッション内のみ有効）
conn = duckdb.connect()

# ファイルベース接続（永続化）
conn_persistent = duckdb.connect('data/analysis.duckdb')

# 基本クエリ
result = conn.execute("""
    SELECT
        floor(longitude / 0.01) AS grid_x,
        floor(latitude  / 0.01) AS grid_y,
        COUNT(*)                AS query_count,
        AVG(is_error::FLOAT)    AS error_rate
    FROM read_parquet('data/llm_geo_logs.parquet')
    GROUP BY grid_x, grid_y
    ORDER BY query_count DESC
    LIMIT 10
""").fetchdf()  # pandas DataFrameとして取得

print(result)

---

2. Polarsの特徴と使い所

2.1 Polarsとは

PolarsはRustで書かれた高速なDataFrameライブラリです。pandasと似たAPIを持ちますが、設計思想が異なります。

2.2 Pandasとの比較

比較軸	Pandas	Polars
内部実装	Python/NumPy（行指向）	Rust（列指向、Arrow）
遅延評価	なし（即時実行）	あり（lazy()で遅延実行）
並列処理	GILにより限定的	マルチコア並列をデフォルトで活用
NULL処理	NaN（浮動小数点）で代替	本物のnull値（Arrow仕様）
適したデータ規模	〜数GB	数GB〜数百GB
メモリ効率	中程度	高い（ゼロコピー変換）

2.3 遅延評価（Lazy Evaluation）

import polars as pl

# 即時実行（Eager）
df_eager = pl.read_parquet('data/llm_geo_logs.parquet')
result_eager = df_eager.filter(pl.col('is_error') == True).group_by('grid_id').agg(pl.len())

# 遅延評価（Lazy）- クエリプランを最適化してから実行
result_lazy = (
    pl.scan_parquet('data/llm_geo_logs.parquet')  # scan_: 遅延読み込み
    .filter(pl.col('is_error') == True)
    .group_by('grid_id')
    .agg(pl.len().alias('error_count'))
    .sort('error_count', descending=True)
    .collect()  # ここで初めて実行される
)

# クエリプランの確認
lazy_frame = (
    pl.scan_parquet('data/llm_geo_logs.parquet')
    .filter(pl.col('is_error') == True)
    .group_by('grid_id')
    .agg(pl.len())
)
print(lazy_frame.explain(optimized=True))  # 最適化後のクエリプランを表示

遅延評価のメリット： - Polarsがクエリ全体を最適化してから実行する（不要な列を読み込まない、フィルタを早期適用する） - 大規模データでメモリ効率が大幅に向上する

2.4 Apache Arrowとの互換性

PolarsはApache Arrowフォーマットを内部で使用します。

# Polars → PyArrow（Arrow形式）
import pyarrow as pa
df_pl = pl.read_parquet('data/llm_geo_logs.parquet')
arrow_table = df_pl.to_arrow()

# Arrow → DuckDB（ゼロコピーで高速）
import duckdb
conn = duckdb.connect()
conn.register('logs_arrow', arrow_table)
result = conn.execute("SELECT COUNT(*) FROM logs_arrow").fetchone()
print(f"レコード数: {result[0]:,}")

---

3. 地理空間拡張（DuckDB spatial）

3.1 spatialのインストールと有効化

import duckdb
conn = duckdb.connect()

# 地理空間拡張のインストール（初回のみ）
conn.execute("INSTALL spatial;")
conn.execute("LOAD spatial;")

print("spatial拡張が有効になりました")

3.2 主要な空間SQL関数

# === ST_Point：点ジオメトリの作成 ===
conn.execute("""
    SELECT ST_Point(141.35, 43.06) AS point_geom
""").show()

# === ST_AsText：ジオメトリをWKT文字列に変換 ===
result = conn.execute("""
    SELECT
        longitude,
        latitude,
        ST_AsText(ST_Point(longitude, latitude)) AS wkt
    FROM read_parquet('data/llm_geo_logs.parquet')
    LIMIT 5
""").pl()
print(result)

# === ST_Distance：2点間の距離（度単位）===
# 距離をメートルに変換するには、度×約111000を使う（緯度方向の近似）
conn.execute("""
    SELECT
        grid_id,
        ST_Distance(
            ST_Point(longitude, latitude),
            ST_Point(141.3544, 43.0618)   -- 札幌駅
        ) * 111000.0 AS dist_from_sapporo_m
    FROM read_parquet('data/llm_geo_logs.parquet')
    LIMIT 10
""").show()

# === ST_Contains：ポリゴン内に点が含まれるか判定 ===
# 例：札幌市中央区のバウンディングボックス内のクエリを抽出
result_central = conn.execute("""
    SELECT COUNT(*) AS count_in_central
    FROM read_parquet('data/llm_geo_logs.parquet')
    WHERE ST_Contains(
        ST_GeomFromText('POLYGON((
            141.31 43.03,
            141.42 43.03,
            141.42 43.10,
            141.31 43.10,
            141.31 43.03
        ))'),
        ST_Point(longitude, latitude)
    )
""").fetchone()
print(f"中央区内のクエリ数: {result_central[0]:,}")

# === ST_GeomFromGeoJSON：GeoJSONからジオメトリ作成 ===
# 例：区境界ポリゴンを読み込んでフィルタリング
conn.execute("""
    SELECT
        a.ward_name,
        COUNT(*) AS query_count,
        AVG(b.is_error::FLOAT) AS error_rate
    FROM read_parquet('data/ward_boundaries.parquet') a
    JOIN read_parquet('data/llm_geo_logs.parquet') b
      ON ST_Contains(
           ST_GeomFromWKB(a.geometry_wkb),
           ST_Point(b.longitude, b.latitude)
         )
    GROUP BY a.ward_name
    ORDER BY query_count DESC
""").show()

3.3 最近傍探索

# 各クエリポイントから最寄り駅までの距離を計算
# （クロスジョイン + 最小値でシミュレート）
result_nearest = conn.execute("""
    WITH logs AS (
        SELECT
            grid_id,
            AVG(longitude) AS lon,
            AVG(latitude)  AS lat
        FROM read_parquet('data/llm_geo_logs.parquet')
        GROUP BY grid_id
    ),
    stations AS (
        SELECT station_name, longitude AS s_lon, latitude AS s_lat
        FROM read_parquet('data/train_stations.parquet')
    ),
    distances AS (
        SELECT
            l.grid_id,
            s.station_name,
            ST_Distance(
                ST_Point(l.lon, l.lat),
                ST_Point(s.s_lon, s.s_lat)
            ) * 111000.0 AS dist_m,
            ROW_NUMBER() OVER (
                PARTITION BY l.grid_id
                ORDER BY ST_Distance(
                    ST_Point(l.lon, l.lat),
                    ST_Point(s.s_lon, s.s_lat)
                )
            ) AS rank
        FROM logs l CROSS JOIN stations s
    )
    SELECT grid_id, station_name, dist_m
    FROM distances
    WHERE rank = 1
    ORDER BY dist_m DESC
    LIMIT 20
""").pl()

print("最寄り駅が遠いグリッドセルTOP20:")
print(result_nearest)

---

4. DuckDB + Polars の連携パターン

4.1 DuckDB → Polars（fetchdf() vs .pl()）

import duckdb
import polars as pl

conn = duckdb.connect()
conn.execute("LOAD spatial;")

# パターン1：.pl()でPolarsとして取得（推奨）
df_polars = conn.execute("""
    SELECT
        date_trunc('day', timestamp) AS day_bin,
        grid_id,
        COUNT(*) AS query_count
    FROM read_parquet('data/llm_geo_logs.parquet')
    GROUP BY ALL
    ORDER BY day_bin, grid_id
""").pl()

print(type(df_polars))  # polars.DataFrame
print(df_polars.head())

# パターン2：.fetchdf()でPandasとして取得
df_pandas = conn.execute("SELECT ...").fetchdf()

# パターン3：Arrowテーブルとして取得してPolarsに変換
df_arrow = conn.execute("SELECT ...").arrow()
df_from_arrow = pl.from_arrow(df_arrow)  # ゼロコピー変換

4.2 Polars → DuckDB（PolarsのDataFrameをSQLで操作）

# Polarsで前処理したDataFrameをDuckDBに渡す
df_preprocessed = (
    pl.scan_parquet('data/llm_geo_logs.parquet')
    .with_columns([
        pl.col('timestamp').dt.truncate('1h').alias('hour_bin'),
        (pl.col('longitude') / 0.01).floor().cast(pl.Int32).alias('grid_x'),
        (pl.col('latitude')  / 0.01).floor().cast(pl.Int32).alias('grid_y'),
    ])
    .collect()
)

# DuckDBにPolars DataFrameを登録
conn.register('preprocessed', df_preprocessed)

# SQLで集計
result = conn.execute("""
    SELECT
        grid_x, grid_y,
        AVG(is_error::FLOAT) AS error_rate,
        STDDEV(response_time_ms) AS response_time_std
    FROM preprocessed
    GROUP BY grid_x, grid_y
    HAVING COUNT(*) >= 10  -- サンプル数が少ないグリッドを除外
""").pl()

print(result.head())

4.3 推奨される連携パターン

大規模Parquet ─→ DuckDB（SQL集計・空間フィルタ）─→ Polars（変換・計算）─→ 機械学習
                      ↑                                    ↓
                 .pl()で変換                        .write_parquet()

---

5. LLMログとの統合処理

5.1 完全なデータパイプライン

import duckdb
import polars as pl
import pandas as pd
from sklearn.preprocessing import StandardScaler
import time

conn = duckdb.connect()
conn.execute("LOAD spatial;")

# ==============================
# ステップ1: DuckDBで大規模集計
# ==============================
t0 = time.perf_counter()

raw_features = conn.execute("""
    WITH log_data AS (
        SELECT
            floor(longitude / 0.01)::INT AS grid_x,
            floor(latitude  / 0.01)::INT AS grid_y,
            timestamp,
            response_time_ms,
            is_error,
            EXTRACT(hour FROM timestamp) AS hour_of_day,
            EXTRACT(dow  FROM timestamp) AS day_of_week   -- 0=日曜
        FROM read_parquet('data/llm_geo_logs.parquet')
        WHERE timestamp >= '2024-01-01'
          AND timestamp <  '2024-04-01'
    )
    SELECT
        grid_x,
        grid_y,
        COUNT(*)                                     AS total_queries,
        SUM(is_error::INT)                           AS total_errors,
        AVG(is_error::FLOAT)                         AS error_rate,
        AVG(response_time_ms)                        AS mean_response_ms,
        PERCENTILE_CONT(0.95) WITHIN GROUP
            (ORDER BY response_time_ms)              AS p95_response_ms,
        STDDEV(response_time_ms)                     AS std_response_ms,
        -- 時間帯別の割合
        AVG(CASE WHEN hour_of_day BETWEEN 9 AND 18
                 THEN 1 ELSE 0 END::FLOAT)           AS daytime_ratio,
        AVG(CASE WHEN day_of_week IN (0, 6)
                 THEN 1 ELSE 0 END::FLOAT)           AS weekend_ratio
    FROM log_data
    GROUP BY grid_x, grid_y
    HAVING COUNT(*) >= 20   -- データが少なすぎるグリッドを除外
    ORDER BY grid_x, grid_y
""").pl()

t1 = time.perf_counter()
print(f"DuckDB集計: {t1-t0:.2f}秒")
print(f"集計結果: {len(raw_features):,}グリッドセル")

# ==============================
# ステップ2: Polarsで特徴量エンジニアリング
# ==============================
features = (
    raw_features
    .with_columns([
        # 対数変換（スキュー補正）
        pl.col('total_queries').log(base=10).alias('log_query_count'),
        pl.col('mean_response_ms').log(base=10).alias('log_mean_response_ms'),

        # エラー数を人口密度で補正
        (pl.col('total_errors') / pl.col('total_queries')).alias('error_per_query'),

        # グリッドID（文字列）
        (pl.col('grid_x').cast(pl.Utf8) + '_' +
         pl.col('grid_y').cast(pl.Utf8)).alias('grid_id'),
    ])
    .drop_nulls()
)

t2 = time.perf_counter()
print(f"Polars特徴量エンジニアリング: {t2-t1:.2f}秒")

# ==============================
# ステップ3: Parquetに永続化
# ==============================
features.write_parquet('data/spatial_features_integrated.parquet',
                        compression='zstd')

t3 = time.perf_counter()
print(f"Parquet保存: {t3-t2:.2f}秒")
print(f"合計処理時間: {t3-t0:.2f}秒")

---

6. ハンズオン手順

6.1 環境のセットアップ

# requirements.txtからインストール
pip install -r requirements.txt

# 確認
python -c "
import duckdb, polars, pyarrow
print(f'DuckDB: {duckdb.__version__}')
print(f'Polars: {polars.__version__}')
print(f'PyArrow: {pyarrow.__version__}')
"

# DuckDB spatialの動作確認
python -c "
import duckdb
conn = duckdb.connect()
conn.execute('INSTALL spatial; LOAD spatial;')
print(conn.execute(\"SELECT ST_AsText(ST_Point(141.35, 43.06))\").fetchone()[0])
"

6.2 Makefile への組み込み

プロジェクトの再現性を高めるために、Makefileでパイプラインを定義します。

# Makefile

.PHONY: all clean data features

# デフォルトターゲット
all: data features

# データの前処理（DuckDB）
data/cube_daily.parquet: data/llm_geo_logs.parquet
    python src/01_build_cube.py

# 特徴量エンジニアリング（DuckDB + Polars）
data/spatial_features_integrated.parquet: data/cube_daily.parquet
    python src/02_feature_engineering.py

features: data/spatial_features_integrated.parquet

# クリーンアップ
clean:
    rm -f data/cube_daily.parquet
    rm -f data/spatial_features_integrated.parquet
    rm -f data/analysis.duckdb
    rm -f output/*.png

実行：

make all        # パイプライン全体を実行
make data       # データ構築のみ
make clean      # 中間ファイルを削除してやり直し

6.3 ベンチマーク比較

import pandas as pd
import polars as pl
import duckdb
import time

DATA_PATH = 'data/llm_geo_logs.parquet'

# --- Pandas ---
t0 = time.perf_counter()
df_pd = pd.read_parquet(DATA_PATH)
result_pd = (
    df_pd
    .assign(hour_bin=df_pd['timestamp'].dt.floor('h'))
    .groupby(['hour_bin', 'grid_id'])
    .agg(
        query_count=('is_error', 'count'),
        error_rate=('is_error', 'mean'),
        mean_response_ms=('response_time_ms', 'mean')
    )
    .reset_index()
)
t_pd = time.perf_counter() - t0

# --- Polars ---
t0 = time.perf_counter()
result_pl = (
    pl.scan_parquet(DATA_PATH)
    .with_columns(pl.col('timestamp').dt.truncate('1h').alias('hour_bin'))
    .group_by(['hour_bin', 'grid_id'])
    .agg([
        pl.len().alias('query_count'),
        pl.col('is_error').mean().alias('error_rate'),
        pl.col('response_time_ms').mean().alias('mean_response_ms'),
    ])
    .collect()
)
t_pl = time.perf_counter() - t0

# --- DuckDB ---
conn = duckdb.connect()
t0 = time.perf_counter()
result_dk = conn.execute(f"""
    SELECT
        date_trunc('hour', timestamp)  AS hour_bin,
        grid_id,
        COUNT(*)                       AS query_count,
        AVG(is_error::FLOAT)           AS error_rate,
        AVG(response_time_ms)          AS mean_response_ms
    FROM read_parquet('{DATA_PATH}')
    GROUP BY hour_bin, grid_id
""").pl()
t_dk = time.perf_counter() - t0

print("=== ベンチマーク結果（時間帯×グリッド集計）===")
print(f"Pandas:  {t_pd:.2f}秒")
print(f"Polars:  {t_pl:.2f}秒  ({t_pd/t_pl:.1f}x faster)")
print(f"DuckDB:  {t_dk:.2f}秒  ({t_pd/t_dk:.1f}x faster)")
print(f"\n結果行数 - Pandas:{len(result_pd):,}, Polars:{len(result_pl):,}, DuckDB:{len(result_dk):,}")

---

7. 演習課題

課題1（基本）

1. DuckDBでdata/llm_geo_logs.parquetから「月別」×「グリッドセル別」のクエリ数とエラー率を集計する（SQLを書くこと）
2. 結果を.pl()でPolarsとして取得し、エラー率が最高の月を特定する

課題2（応用）

1. DuckDBのST_Distanceを使って、各グリッドセルから「札幌駅（141.3544, 43.0618）」までの距離を計算する
2. 距離帯（0-1km, 1-5km, 5-20km, 20km以上）別のエラー率をPolarsで集計し、棒グラフを描く

課題3（発展）

1. Pandas, Polars, DuckDBのベンチマークを自分のデータで実行し、結果を比較する
2. Makefileを作成して、以下の処理をmake all一発で再現できるようにする：
3. data/spatial_features_integrated.parquetの生成
4. output/cluster_map.pngの生成（第11回のk-meansクラスタリング結果）
5. README.mdに再現手順を記述する（第14回の最終プロジェクトに向けた準備）

提出形式：Jupyterノートブック＋Makefile、ファイル名：13_duckdb_polars_[学籍番号].ipynb

---

まとめ

項目	ポイント
DuckDB	インプロセスOLAP、Parquetを直接SQLで処理、空間拡張でGIS機能も使える
Polars	Rust製・列指向・遅延評価・マルチコア並列、大規模データで高速
連携の基本	DuckDB→.pl()でPolars DataFrame取得、Polars→conn.register()でSQL利用
空間SQL	ST_Contains, ST_Distance, ST_AsText, ST_GeomFromTextを組み合わせる
Parquet永続化	write_parquet(compression='zstd')で中間結果を保存し、Makefileで管理
速度感	PandasのGROUP BY集計に対してPolarsは3〜10倍、DuckDBはさらに速いケースも
次回への接続	第14回では本日構築したパイプラインを再現リポジトリとしてまとめ、発表する

次回予告： 第14回（最終回）は、全講義の成果物を統合した再現リポジトリを発表します。リポジトリ要件を事前に確認しておいてください。

---

作成：小野原彩香 / ビッグデータ応用Ⅱ〔特徴量エンジニアリング〕