如何将Spark SQL转换为DataFrame API并优化大表Join性能

对应SQL逻辑的DataFrame写法如下，提前做谓词过滤减少参与计算的数据量：

from pyspark.sql import functions as F

# 提前过滤无效数据，利用谓词下推减少Join阶段数据量
trans_valid = transactions.filter(F.col("is_blocked") == False)
users_valid = users.filter(F.col("is_active") == 1)

# 关联、聚合、排序逻辑
res = trans_valid.join(
        users_valid,
        on="user_id",
        how="inner"
    )\
    .groupBy("transaction_category_id")\
    .agg(
        F.sum("transaction_amount").alias("sum_amount"),
        F.countDistinct("user_id").alias("num_users")
    )\
    .orderBy(F.col("sum_amount").desc())

res.show()

如果要强制触发广播Join，只需要把Join语句里的小表用F.broadcast()包裹即可，写法为trans_valid.join(F.broadcast(users_valid), on="user_id", how="inner")。

广播Join（Broadcast Join）适用 #

• 你的场景是典型的大小表关联：交易表transactions是大表，用户表users是小表，完全符合广播Join的适用条件。
• 广播Join的原理是把过滤后的小表全量分发到所有计算节点，全程不需要对大表做Shuffle，能省掉大表Shuffle带来的网络传输、磁盘读写开销，性能提升非常明显。
• 注意事项：Spark默认自动广播小于10MB的表，如果你过滤后的有效用户表大小在1GB以内（可根据集群Executor内存配置适当放宽），可以调大spark.sql.autoBroadcastJoinThreshold参数适配表大小，或者手动加F.broadcast()标记强制广播小表即可。

加盐（Salting）技术不适用 #

• 加盐技术的作用是解决两个大表Shuffle Join时出现的数据倾斜问题：通过给倾斜的Join键加随机前缀，把原本集中在单个Task上的倾斜数据打散到多个Task处理。
• 你的场景如果用了广播Join，Join阶段完全不会产生Shuffle，根本不存在Join阶段的数据倾斜问题，不需要加盐。只有当用户表规模也大到无法广播，必须走Shuffle Join，且确认user_id键存在严重数据倾斜（比如大量空值、单个测试用户对应超大量交易）时，才需要考虑用加盐方案，否则加了反而会增加额外计算开销。

内容的提问来源于stack exchange，提问作者Durga