バッチ処理オプティマイザー

バッチ処理オプティマイザー は何をしますか？

Spark、Polars、DuckDB、pandasを使用したバッチデータ処理のワークロードを、メモリ効率・並列性・コスト削減に焦点を当てて最適化します。

起動トリガー

アクティブ化条件: 「バッチ処理」「Spark最適化」「Polars」「DuckDB」「pandasパフォーマンス」「データフレーム」「シャッフル最適化」「パーティションスキュー」「メモリ最適化」「メモリ不足」

対象外: リアルタイムストリーミング → streaming-pipeline-architect | ウェアハウスSQLチューニング → data-warehouse-optimizer | パイプラインオーケストレーション → airflow-dag-orchestrator

クイックスタート

1. 適切なツールを選ぶ — 単一ノード分析にはDuckDB、DataFrameにはPolars、分散処理にはSpark
2. まずプロファイリング — 最適化前にボトルネック（シャッフル、スキュー、メモリ）を特定する
3. 早期にデータを削減 — パイプラインのできるだけ早い段階でフィルタリングとカラム選択を行う
4. シャッフルを避ける — 小さなテーブルはブロードキャストし、データを事前パーティション化、マップサイド結合を使用する
5. 適切なリソースサイズ — エグゼキュータのメモリ/コアを実際のデータサイズに合わせる

コア機能

ドメイン	テクノロジー
分散処理	Apache Spark 3.5+, Dask, Ray
単一ノード	DuckDB 1.1+, Polars 1.x, pandas 2.2+
ファイル形式	Parquet, Arrow IPC, Delta Lake, Iceberg
最適化	AQE (Spark), 遅延評価 (Polars), カラムナー型スキャン
クラウド	Databricks, EMR, Dataproc, サーバーレスSpark

アーキテクチャパターン

ツール選択の決定木

データサイズ?
  ├─ < 10 GB     → DuckDB (SQL) または Polars (DataFrame)
  │                 単一マシン、ゼロセットアップ、最速のイテレーション
  │
  ├─ 10-100 GB   → Polars (遅延) または DuckDB (アウトオブコア)
  │                 依然として単一マシン、ディスクへのスピルが可能
  │
  └─ > 100 GB    → Spark (分散処理)
                    マルチノードクラスター、シャッフルベースの結合

複雑さ?
  ├─ SQL中心     → DuckDB (分析向け最速のSQLエンジン)
  ├─ DataFrame    → Polars (pandasより10倍高速、遅延評価)
  └─ 複雑なML    → Spark + MLlib または Spark + Ray

Spark最適化パターン

from pyspark.sql import SparkSession
import pyspark.sql.functions as F

spark = SparkSession.builder \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .getOrCreate()

# 良い例: 小さなディメンションテーブル（< 100MB）をブロードキャスト
from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_dim_df), "key")

# 良い例: 述語プッシュダウン — 結合前にフィルタリング
orders = spark.read.parquet("s3://data/orders/") \
    .filter(F.col("order_date") >= "2026-01-01") \
    .select("order_id", "customer_id", "amount")  # カラムプルーニング

# 悪い例: 大規模データセットでのcollect() — ドライバーでOOMを引き起こす
# all_data = large_df.collect()  # 絶対にやらない

# 良い例: パーティション化された出力を書き込む
result.repartition(200) \
    .write.mode("overwrite") \
    .partitionBy("order_date") \
    .parquet("s3://output/results/")

Polars遅延評価

import polars as pl

# 遅延モード: クエリプランを構築し、最適化してから実行
result = (
    pl.scan_parquet("data/orders/*.parquet")  # 遅延スキャン
    .filter(pl.col("order_date") >= "2026-01-01")
    .join(
        pl.scan_parquet("data/customers/*.parquet"),
        on="customer_id",
        how="inner"
    )
    .group_by("region")
    .agg([
        pl.col("amount").sum().alias("total_revenue"),
        pl.col("order_id").n_unique().alias("order_count"),
    ])
    .sort("total_revenue", descending=True)
    .collect()  # 最適化されたプランを実行
)

# Polarsは自動的に: 述語プッシュダウン、射影プッシュダウン、
# 結合の並べ替えを遅延評価で最適化

アンチパターン

1. 5GB超でのpandas — pandasは全データをメモリに読み込む; 中規模データにはPolars（遅延）またはDuckDBを、大規模データにはSparkを使用する
2. ドライバーへのcollect — 大規模なSpark DataFrameでのdf.collect()やdf.toPandas()はOOMを引き起こす; 事前に集約する
3. パーティションスキューの無視 — 一つのパーティションに他の10倍のデータがあるとジョブ全体がボトルネックになる; AQEまたはソルティングを使用する
4. 全カラムの読み取り — 必要なカラムのみを常に選択する; Parquetのカラムナー形式は未使用のカラムを完全にスキップする
5. 小さすぎる出力ファイル — 小さな出力ファイル（< 128MB）が多すぎると後続の読み取りが遅くなる; 書き込み前にcoalesceする

品質チェックリスト

• ツールがデータサイズに一致している（DuckDB/Polars < 100GB, Spark > 100GB）
• カラムが早期にプルーニングされている（必要なものだけを選択）
• フィルタがスキャンレベルでプッシュダウンされている（述語プッシュダウン）
• 小さなテーブルが結合時にブロードキャストされている（< 100MB）
• Spark AQEが有効（適応型クエリ実行）
• 大規模データセットでcollect()を使用していない（事前に集約してから収集）
• 出力ファイルのサイズが128MB〜1GBに設定されている（書き込み前にcoalesce/repartition）
• パーティションスキューが監視・緩和されている（ソルティングまたはAQE）
• ジョブがプロファイリングされている: Spark UIのステージ、Polarsの.explain()、DuckDBのEXPLAIN ANALYZE
• メモリが適切に設定されている: エグゼキュータメモリ >= 最大パーティションサイズの2倍