别再把 Spark / Dask 当“放大版 Pandas”了

——聊聊大规模特征计算那些真能救命的技巧

说实话，这几年我见过太多团队，明明上了 Spark / Dask，特征计算却还是慢得想骂人。
任务一跑就是几个小时，CPU 在抖，内存在炸，工程师在群里装死。

然后大家开始甩锅：

• “Spark 不适合做特征工程”
• “Dask 不稳定”
• “是不是该上 Flink 了？”
• “要不直接换 ClickHouse？”

我一般会很冷静地回一句：

兄弟，不是框架不行，是你把它当成了 Pandas。

今天这篇文章，我不想讲教科书那套 API，我想讲点真正踩过坑、救过命的经验：
怎么用 Spark / Dask，老老实实把「大规模特征计算」这件事干好。

一、先把话说明白：特征计算的本质不是“算”，是“搬 + 聚”

很多人一上来就写：

df.groupBy("user_id").agg(
    F.avg("click_cnt"),
    F.max("stay_time"),
    F.count("*")
)

然后一跑：Shuffle 爆炸，任务拖到天荒地老。

为啥？

因为大规模特征计算 ≈ 数据重分布 + 状态聚合，而不是你脑子里那点数学公式。

我一直有个很“土”的认知模型：

Spark / Dask 80% 的时间，都花在数据怎么动上，而不是怎么算。

所以第一条铁律是：

👉 能不 shuffle，就别 shuffle

二、Spark：特征工程跑得慢，90% 都死在 Shuffle 上

1️⃣ 小表 Join，别傻乎乎地 Join

这是 Spark 特征计算里最容易被忽视、但收益最大的优化点。

错误示范：

features = big_df.join(user_profile_df, "user_id")

如果 user_profile_df 只有几百万行，你这一步等于把小表反复拷贝、反复 shuffle。

正确姿势：广播 Join

from pyspark.sql.functions import broadcast

features = big_df.join(
    broadcast(user_profile_df),
    on="user_id",
    how="left"
)

💡 我的经验是：

只要能广播，就一定广播。
广播不是“优化技巧”，是“工程常识”。

2️⃣ 特征计算，先 Repartition 再 GroupBy

很多人不理解这句，但我可以很负责任地说：

80% 的 Spark GroupBy 慢，是分区策略错了。

错误写法：

df.groupBy("user_id").agg(F.sum("cnt"))

Spark 会临时做一次全局 shuffle。

更稳的写法：

df = df.repartition(200, "user_id")

features = df.groupBy("user_id").agg(
    F.sum("cnt").alias("cnt_sum")
)

这一步不是“多此一举”，而是提前告诉 Spark：你要按什么维度分桶。

📌 实战经验：

• 特征 key 是 user_id / item_id → 一定提前 repartition
• 分区数宁多勿少（后面还能 coalesce）

3️⃣ 能用内置函数，别碰 UDF（真的）

UDF 在特征工程里，属于慢性自杀。

错误示范：

@udf("double")
def ratio(a, b):
    return a / (b + 1e-6)

正确示范：

from pyspark.sql.functions import col

df = df.withColumn("ratio", col("a") / (col("b") + 1e-6))

原因很简单：

• UDF = JVM ↔ Python 频繁切换
• Catalyst 优化器直接失效

我见过一个项目，删掉 3 个 UDF，任务时间从 40 分钟掉到 6 分钟。

三、Dask：别把它当“Spark 平替”，它是另一种生物

很多 Python 团队用 Dask，是因为一句话：

“我们只会 Pandas。”

这句话既是 Dask 的优势，也是它最大的坑。

1️⃣ Dask 特征工程，先想“图”，再想“代码”

Dask 不是马上算，它是：

先建任务图（Task Graph），再一次性执行。

所以写法顺序很重要。

推荐模式：

import dask.dataframe as dd

df = dd.read_parquet("events.parquet")

features = (
    df.groupby("user_id")
      .agg({
          "click": "sum",
          "stay_time": "mean"
      })
)

result = features.compute()

🚨 千万别这样写：

df = df.compute()
# 然后再 groupby

这等于：我先把所有数据拉到单机，再假装自己是大数据工程师。

2️⃣ 分区大小，决定 Dask 生死

Dask 官方说过一句非常真实的话：

Too many small tasks are worse than a few big ones.

我的经验参数：

• 每个 partition：100MB～300MB
• 特征计算阶段：减少 partition 数量

df = df.repartition(partition_size="200MB")

这一步对稳定性和性能提升都非常明显。

3️⃣ 特征 Join：先对齐分区，再 Join

Dask 的 Join，如果分区不一致，会非常痛苦。

df1 = df1.set_index("user_id")
df2 = df2.set_index("user_id")

features = df1.join(df2)

📌 这一步的本质是：

我宁愿现在慢一点做一次 index 对齐，也不愿意之后每一步都在乱跑。

四、一个我非常真实的观点：

特征工程不是“算力问题”，是“工程取舍问题”

我见过太多团队：

• 一边骂 Spark 慢
• 一边一天跑 20 次全量特征
• 一边所有特征都不设 TTL
• 一边 key 设计得跟艺术品一样复杂

我现在的原则非常简单：

不是所有特征，都配得上“每天全量重算”。

一些非常实用的策略：

• 时间衰减特征 → 增量算
• 用户静态属性 → 离线算一次，缓存
• 长窗口统计 → 周级 / 月级算
• 探索性特征 → 小样本先验证

Spark / Dask 只是工具，真正决定效率的，是你对业务节奏的理解。

五、写在最后：工具会过时，但“算得明白”不会

说句掏心窝子的话：

会 Spark / Dask 的人很多，
会用它们把特征工程“算明白”的人很少。

真正厉害的工程师，不是 API 背得多，而是：

• 知道哪里该重算
• 知道哪里该缓存
• 知道哪一步在浪费 Shuffle
• 知道哪些特征其实没业务价值