curiositech/windags-skills

批量处理优化器

使用 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 join
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, serverless 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 或加盐
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 倍