PySpark大数据量下基于ID范围过滤DataFrame的高效方法问询

针对df1（1亿条）、df2（超10亿条）基于ID+数值范围的过滤场景，原left semi join方案会因大量shuffle和跨节点匹配导致性能瓶颈，以下是几种更高效的实现方式：

1. 分区对齐+分桶优化 #

核心思路是让相同ID的数据落在同一节点/桶内，避免全量shuffle，仅在桶内做范围匹配：

• 先按id对两个DataFrame做相同数量的分区/分桶，确保同ID数据被分配到同一executor
• 分桶数需根据集群资源（executor数量、内存）调整，建议设置为executor数量的2-4倍

# 按id重分区，示例设置1000个分区（根据集群规模调整）
df1_part = df1.repartition(1000, "id")
df2_part = df2.repartition(1000, "id")

# 执行桶内left semi join
join_cond = [
    df1_part.id == df2_part.id,
    df1_part.start_dt_int <= df2_part.enc_dt_int,
    df2_part.enc_dt_int <= df1_part.end_dt_int
]
result = df2_part.join(df1_part, on=join_cond, how="leftsemi")
result.show()

2. 广播ID范围字典（适用于ID基数较小场景） #

若每个ID对应的范围数量不多，可将df1的范围数据转换为字典并广播，直接在df2的map阶段完成过滤，完全避免join操作：

• 注意：若ID基数接近1亿，广播字典会占用大量内存，仅适合ID基数远小于集群内存承载能力的场景

from pyspark.sql import functions as F

# 按ID聚合所有对应的数值范围
df1_ranges = df1.groupBy("id").agg(
    F.collect_list(F.struct("start_dt_int", "end_dt_int")).alias("ranges")
)
# 转换为本地字典并广播
id_range_map = {row.id: row.ranges for row in df1_ranges.collect()}
broadcast_ranges = spark.sparkContext.broadcast(id_range_map)

# 定义过滤函数
def match_range(row):
    ranges = broadcast_ranges.value.get(row.id, [])
    for r in ranges:
        if r.start_dt_int <= row.enc_dt_int <= r.end_dt_int:
            return True
    return False

# 直接过滤df2
result = df2.rdd.filter(match_range).toDF(df2.schema)
result.show()

3. 使用SQL EXISTS子句（利用Catalyst优化器） #

Spark SQL的Catalyst优化器对EXISTS子句有特殊优化，相比DataFrame API的join，能更高效地裁剪不必要的数据：

# 注册临时视图
df1.createOrReplaceTempView("df1")
df2.createOrReplaceTempView("df2")

# 执行存在性查询
result = spark.sql("""
    SELECT * FROM df2
    WHERE EXISTS (
        SELECT 1 FROM df1
        WHERE df1.id = df2.id
        AND df1.start_dt_int <= df2.enc_dt_int
        AND df2.enc_dt_int <= df1.end_dt_int
    )
""")
result.show()

4. 热点ID单独处理（解决数据倾斜） #

若存在少数ID对应的df2数据量极大（热点ID），会导致单个executor负载过高，此时可拆分处理：

1. 统计df2的ID分布，找出热点ID
2. 对热点ID，单独广播其对应的df1范围，过滤df2中的热点数据
3. 非热点ID使用分桶join处理
4. 最后合并结果

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