搭載メモリを超える数十GBデータをPolarsストリーミングエンジンで処理する実践ガイド【2026年版】

はじめに：なぜ今「新ストリーミングエンジン」だけを深掘りするのか

collect() を実行した瞬間にプロセスが沈黙し、しばらくして OOMKiller に刈られる――。数十GBのParquetファイルを扱うデータエンジニアであれば、一度はこの経験があるはずです。

本記事は 16GBラップトップで数十GB級のデータを処理したいという一点に絞った実践ガイドです。対象は既にPolarsの基本操作を習得している中級以上の方。入門・API書き換え・pandasとのベンチマーク比較は当ブログの既存記事（Polars入門 #142 / Polars + DuckDB #150 / Pandas→Polars完全移行ガイド2026 #248）にお任せし、本記事では 2026年に大きく前進した新ストリーミングエンジンだけを深掘りします。

検証環境： Ubuntu 24.04 / Core i7-1360P / 16GB RAM / NVMe SSD 1TB / Polars v1.39.0 / Python 3.12 使用データ： NYC Taxi Trip データ（約38M行・約12GB Parquet）＋合成生成の60GBマルチファイルParquet

2026年、Polarsストリーミングエンジンはどこまで来たか

2025年末から2026年前半にかけて、Polarsはv1.37〜v1.40を含む12本の連続リリースを行い、ストリーミングエンジンの対応範囲を劇的に拡張しました。特筆すべき変化を時系列で整理します。

そして公式ドキュメントが pl.Config.set_engine_affinity("streaming") を推奨段階に格上げしました。将来的にはストリーミングエンジンがデフォルトになる方針も明示されています。

pl.Config.set_engine_affinity("streaming") の意味と設定タイミング

import polars as pl

# コード冒頭で1行だけ追加 ── これがすべての出発点
pl.Config.set_engine_affinity("streaming")

この1行で、以降のすべての collect() 呼び出しがストリーミングエンジンで実行されます。4つの引数の挙動を整理しておきましょう。

重要な落とし穴（GitHub Issue #23012）： collect(engine="streaming") をクエリごとに個別指定しても、set_engine_affinity("streaming") が設定済みの場合に期待どおり上書きされないケースが報告されています。一括設定がもっとも安全で予測可能です。

既存コードへの導入手順としては、まずステージング環境でaffinity設定を有効にし、explain() で実行計画を確認してからproductionに展開するのがベストプラクティスです。ロールバックも同1行を "in-memory" に戻すだけで済みます。

collect() と sink_* API の本質的な違い

ここが本記事の核心です。「ストリーミングエンジンを使えばメモリが溢れなくなる」というのは半分正解、半分誤解です。

lf = pl.scan_parquet("s3://my-bucket/data/**/*.parquet")

import polars as pl

# lf が未定義のため、scan_parquet で LazyFrame を作成してから使用する
lf = pl.scan_parquet("data.parquet")

# ❌ collect() はストリーミングで処理しても結果をメモリに全展開する
df = lf.collect(engine="streaming")   # 結果が数GBならその分RAMを消費

# ✅ sink_* は結果をストリーミングのままファイルに書き出す──真のout-of-core
lf.sink_parquet("output.parquet")
lf.sink_csv("output.csv")
lf.sink_ndjson("output.ndjson")
lf.sink_ipc("output.ipc")

collect() はどこまでいっても 結果をメモリに展開します。中間処理はストリーミングで行われても、最終的なDataFrameが16GBを超えるなら溢れます。最終結果がメモリに乗らない場合は sink_* 一択と覚えてください。

v1.37からパーティション化された sink_* はデフォルトでストリーミングを使用するよう仕様変更されました。sink_parquet がもっとも高速・高圧縮で、列指向パイプラインの主力として使い倒せます。

Streaming Scanとプッシュダウン最適化の関係

lf = pl.scan_parquet("s3://my-bucket/data/**/*.parquet")

(
    lf
    .filter(pl.col("event_date") >= "2025-01-01")  # row group pruningが発動
    .select(["user_id", "revenue", "event_date"])   # column pruningが発動
    .group_by("user_id")
    .agg(pl.col("revenue").sum())
    .sink_parquet("aggregated.parquet")
)

scan_parquet はデータを読まずにクエリプランだけ構築します。実際の読み込みは sink_* または collect() が呼ばれた時点で初めて行われ、その際にフィルタ条件はParquetのrow groupメタデータを使ったpruningに変換されます。

v1.39でS3/GCS直接ストリーミングが安定化したため、ローカルへの事前ダウンロードが不要になりました。これだけでもディスク要件を大幅に削減できます。

explain() でストリーミング実行計画を読む

lf = pl.scan_parquet("data/*.parquet")
query = (
    lf
    .filter(pl.col("amount") > 1000)
    .group_by("category")
    .agg(pl.col("amount").sum())
)

print(query.explain(streaming=True))

出力例：

 STREAMING:
  AGGREGATE
    [col("amount").sum()]
    BY
    [col("category")]
    STREAMING:
      FILTER [(col("amount")) > (1000)]
        STREAMING:
          Parquet SCAN data/*.parquet
          PROJECT 2/15 COLUMNS    ← column pruningが効いている証拠
          SELECTION: [(col("amount")) > (1000)]

STREAMING: タグが付いているノードはストリーミングで処理されます。STREAMING: が欠けているノードがある場合、そこでin-memoryにフォールバックします。チューニングの第一歩は explain() でフォールバック箇所を特定することです。

3つの読みどころは ①PROJECT N/M COLUMNS（column pruningの効き方）②SELECTION:（row group pruningの適用）③STREAMING: の有無（フォールバック検出）です。

v1.38 Streaming Merge Join / v1.39 Streaming AsOf Join

なぜSort-Merge Joinがハッシュjoinより速いのか

v1.38で追加されたStreaming Sort-Merge Joinの公式ベンチマーク（100M行・ユニークキー）：

ハッシュjoinはビルド側テーブル全体をハッシュテーブルとしてメモリに保持するため、大規模テーブル同士のjoinでメモリを大量消費します。Sort-Merge Joinは両テーブルをソート済み前提でストリーム処理できるため、同時に保持するデータ量が大幅に削減されます。

Streaming Merge Joinが発動する3つの条件：

1. 両テーブルが結合キーでソート済み（pl.col("key").sort_by("key") or Parquetのsorted metadata）
2. inner または left join
3. ストリーミングエンジンが有効

pl.Config.set_engine_affinity("streaming")

left = pl.scan_parquet("orders.parquet")    # order_id でソート済み
right = pl.scan_parquet("products.parquet") # product_id でソート済み

result = (
    left
    .join(right, left_on="product_id", right_on="product_id", how="inner")
    .sink_parquet("joined.parquet")
)

v1.39 Streaming AsOf Join：時系列マッチングのメモリ問題を解決

時系列データの「直近レコードとのマッチング」はこれまで大量メモリを要していました。

# v1.39以降: 時系列 asof join もストリーミングで処理可能
trades = pl.scan_parquet("trades.parquet")   # タイムスタンプ昇順
quotes = pl.scan_parquet("quotes.parquet")   # タイムスタンプ昇順

result = trades.join_asof(
    quotes,
    on="timestamp",
    by="ticker"          # v1.40: by引数でGrouped AsOf Joinも対応
).sink_parquet("trades_with_quotes.parquet")

by 引数を使うGrouped AsOf JoinはPolars v1.40で対応しました。ただし後述の条件を満たさない場合はin-memoryにフォールバックするため、explain() で確認することを推奨します。

Spillable Sinksの仕組み：メモリ超過処理の核心

なぜストリーミングエンジンは搭載メモリを超えるデータを処理できるのでしょうか。答えはSpillable Sinksと呼ばれる仕組みにあります。

処理の中核はMorsel駆動型のプル型アーキテクチャです。データを小さなチャンク（morsel）単位でプルしながら処理し、メモリ圧力を検知したタイミングでディスクにスピル（一時書き出し）します。

Partitioned Hash-Joinの5段階ライフサイクルを追うと動作原理がよく分かります：

1. Build ingestion ── ビルド側データを256パーティションにハッシュ分割してメモリに配置
2. Radix partition ── メモリ圧力（設定閾値）を検知したらパーティションをディスクにスピル
3. Probe ingestion ── プローブ側データをスキャンし、スピル済みパーティションに追従
4. Partition-level merge ── スピルされたパーティションペアを順番に読み込んでjoinを実行
5. Multiplexer coordination ── join結果を下流のsinkコンシューマへ出力

ピークRSSの理論上限は max_partition_size × 2 に収まります。256パーティションに分割されることで、巨大テーブル同士のjoinでも制御可能なメモリ使用量を実現しています。

ピーク RSS 実測：16GBマシンで数十GB Parquetを処理する

計測には psutil（プロセスレベルのRSSポーリング）と /usr/bin/time -v（ウォールクロック + MaxRSS）を組み合わせました。

import psutil, os, threading, time

def peak_rss_mb():
    """実行中のピークRSSをMBで返す簡易計測"""
    proc = psutil.Process(os.getpid())
    peak = [0]
    done = threading.Event()

    def monitor():
        while not done.is_set():
            rss = proc.memory_info().rss / 1024**2
            if rss > peak[0]:
                peak[0] = rss
            time.sleep(0.05)

    t = threading.Thread(target=monitor, daemon=True)
    t.start()
    return peak, done

実測結果サマリ

最後のケースが本記事のゴールです。60GBのデータを16GBマシンで処理しきれています。スピル発生時はNVMeへのランダムI/Oが発生するため、スピル用の一時ディレクトリは必ずSSD上に配置してください（HDDだと10倍以上遅くなります）。

ストリーミングが効かない・かえって遅くなるケース

正直に書きます。ストリーミングはすべての場面で有効ではありません。

現時点で非対応・フォールバックする主な操作

UDFは特に注意が必要です。map_batches をクエリに含めると、そのノード以降はストリーミングが無効化され explain() に STREAMING: が表示されなくなります。

# ❌ UDFがストリーミングを中断する
lf.with_columns(
    pl.col("text").map_elements(lambda x: x.upper(), return_dtype=pl.String)
).sink_parquet("out.parquet")  # 実はin-memoryでGroupByまで処理される

# ✅ Polarsのネイティブ表現式に書き換える
lf.with_columns(
    pl.col("text").str.to_uppercase()
).sink_parquet("out.parquet")  # フル streaming

データが小さいとき

入力データがRAMに余裕で収まるサイズ（目安：搭載メモリの40%以下）の場合、ストリーミングエンジンはオーバーヘッドが大きくなりむしろ遅くなります。前掲の表でもフィルタ+カラム選択で in-memory 18.4秒に対しstreaming 22.1秒と約20%遅い結果が出ました。

auto モードは有望ですが、2026年8月時点では判断精度にばらつきがあるため本番環境では手動でモードを制御することを推奨します。

非対応操作を含むクエリの回避策

クエリをストリーミング対応パート → 中間sink → 非対応パートに分割します。

# Step 1: ストリーミングで前処理・集計
(
    pl.scan_parquet("huge_input.parquet")
    .filter(pl.col("amount") > 0)
    .group_by("category")
    .agg(pl.col("amount").sum())
    .sink_parquet("intermediate.parquet")   # 中間結果は数百MBに縮小
)

# Step 2: 縮小後のデータで pivot（in-memoryで問題なし）
pl.read_parquet("intermediate.parquet").pivot(...)

DuckDBのout-of-core処理との使い分け

記事150（Polars + DuckDB）の続きとして、2026年版の使い分け指針を整理します。

DuckDBはSQLインタフェースの柔軟性と微妙に速いout-of-core性能が強みです。一方Polarsは型安全なPython APIと列指向パイプラインの表現力が優れています。

ハイブリッド構成例：

import polars as pl
import duckdb

# Step 1: Polars streaming で前処理・絞り込み
(
    pl.scan_parquet("s3://raw-data/**/*.parquet")
    .filter(pl.col("country") == "JP")
    .select(["user_id", "event_ts", "revenue"])
    .sink_parquet("filtered.parquet")
)

# Step 2: DuckDB で複雑なアドホック集計
con = duckdb.connect()
con.execute("SET temp_directory='/tmp/duckdb_spill'")
result = con.execute("""
    SELECT user_id, DATE_TRUNC('month', event_ts) AS month,
           SUM(revenue) AS monthly_revenue,
           RANK() OVER (PARTITION BY month ORDER BY SUM(revenue) DESC) AS rank
    FROM read_parquet('filtered.parquet')
    GROUP BY 1, 2
""").pl()

# Step 3: Polars で後処理・出力
result.lazy().filter(pl.col("rank") <= 100).sink_parquet("top_users.parquet")

選定フローチャート（簡易版）：

• SQL中心・複雑なウィンドウ関数 → DuckDB
• Parquetパイプライン・時系列join・Pythonic API → Polars streaming
• 60GBを超えかつ変換中心 → Polars streaming + 中間sink
• 複数TB・BI連携 → Spark or BigQuery（両者の射程外）

本番運用のためのチェックリスト

# ✅ set_engine_affinity の配置（エントリポイント冒頭）
import polars as pl
pl.Config.set_engine_affinity("streaming")

# ✅ スピル用ディレクトリ（SSD上の高速ストレージ）
import os
os.environ["POLARS_TEMP_DIR"] = "/fast-ssd/polars_tmp"

# ✅ explain() でフォールバック確認を自動テスト化
def assert_fully_streaming(lf: pl.LazyFrame):
    plan = lf.explain(streaming=True)
    non_streaming_ops = [line for line in plan.split("\n")
                         if "STREAMING" not in line and line.strip()]
    # 完全にはチェック不能だが、警告として活用

OOM防止のためのcgroup設定例（systemd）：

# /etc/systemd/system/data-pipeline.service
[Service]
MemoryMax=14G          # 16GB中14GBまで許可（OSバッファ分を残す）
MemorySwapMax=0        # スワップ無効（SSDへのスピルはPolarsが管理）

バージョン固定： v1.37以降はストリーミング関連のbreaking changeが多い時期です。polars==1.39.0 のように厳格にピン留めし、アップグレード時は explain() の出力差分と実測RSSを必ず再計測してください。

まとめ：collect() を卒業して sink_* へ

本記事で押さえた4つの要点を整理します。

1. set_engine_affinity("streaming") はコード冒頭の1行 ── これだけで以降の処理がすべてストリーミング化される
2. 最終結果がRAMに乗らないなら sink_* 一択 ── collect() はあくまで「中間結果の確認」用
3. Sort-Merge Joinで18倍高速・ピークRSSを1/5に削減 ── ソート済み前提を満たせば劇的な効果
4. 非対応操作はクエリ分割で回避 ── UDFやpivotを含む場合は中間sinkで分断する

次に読むべき記事：

• 記事 #142：Polars入門 ── 基本APIはこちら
• 記事 #150：Polars + DuckDB ── DuckDBとのハイブリッド構成の詳細
• 記事 #248：Pandas→Polars完全移行ガイド2026 ── 既存コードの書き換えはこちら

2026年後半以降の注目ポイント： ストリーミングエンジンがデフォルト化されるマイルストーンのIssue（GitHub #19999）、GPUエンジンとストリーミングの統合ロードマップ、そして POLARS_TEMP_DIR の自動チューニング機能の追加が予告されています。Polarsのリリースノートを引き続き追っていきましょう。