curiositech/batch-processing-optimizer

批次處理最佳化工具

最佳化使用 Spark、Polars、DuckDB 與 pandas 的批次資料處理工作負載，專注於記憶體效率、平行處理與成本降低。

觸發條件

啟用於：「批次處理」、「Spark 最佳化」、「Polars」、「DuckDB」、「pandas 效能」、「資料幀」、「shuffle 最佳化」、「分割傾斜」、「記憶體最佳化」、「記憶體不足」

**不適用於：**即時串流 → streaming-pipeline-architect | 資料倉儲 SQL 調校 → data-warehouse-optimizer | 管線編排 → airflow-dag-orchestrator

快速入門

1. 選擇正確的工具 — 單節點分析用 DuckDB，DataFrame 用 Polars，分散式用 Spark
2. 先剖析 — 最佳化前先找出瓶頸（shuffle、傾斜、記憶體）
3. 早期縮減資料 — 在管線中盡早進行過濾與欄位選擇
4. 避免 shuffle — 廣播小表格、預先分區資料、使用 map-side 聯結
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（分散式）
                    多節點叢集、shuffle 式聯結

複雜度？
  ├─ 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. 收集至驅動程式 — 對大型 Spark DataFrame 使用 df.collect() 或 df.toPandas() 會造成 OOM；請先彙總
3. 忽略分割傾斜 — 某個分割承載比其它分割多 10 倍的資料，會拖慢整個工作；請使用 AQE 或加鹽（salting）
4. 讀取所有欄位 — 永遠只選擇需要的欄位；Parquet 欄位格式會完全跳過未使用的欄位
5. 輸出檔案過小 — 太多小型輸出檔案（< 128MB）會拖慢下游讀取；寫入前請先合併（coalesce）

品質檢查清單

• 工具與資料大小匹配（DuckDB/Polars < 100GB，Spark > 100GB）
• 欄位已提前修剪（只選取需要的欄位）
• 過濾條件已下推至掃描層級（述詞下推）
• 聯結中小表格已廣播（< 100MB）
• Spark AQE 已啟用（自適應查詢執行）
• 未對大型資料集使用 collect()（彙總後再收集）
• 輸出檔案大小為 128MB–1GB（寫入前已 coalesce/repartition）
• 分割傾斜已監控並緩解（salting 或 AQE）
• 工作已剖析：Spark UI 階段、Polars .explain()、DuckDB EXPLAIN ANALYZE
• 記憶體配置適當：執行器記憶體 >= 最大分割的 2 倍