Glue/Spark：如何高效过滤带数千条件的大型DynamicFrame

当前实现存在三个致命性能问题：

• 逐行遍历pandas规则表对百万级时序DynamicFrame做过滤，等价于对时序表执行数千次全表扫描，计算和IO开销随规则数线性放大
• 循环union的方式会生成极长的执行计划血缘链，Spark Catalyst优化阶段就会产生极高开销，甚至直接触发OOM
• 过滤逻辑用Python lambda实现，数据需要逐行在JVM和Python进程间序列化传输，完全没有利用Spark原生向量化计算能力

核心思路是将仅数千行的pandas规则表转换为Spark/Glue原生分布式表，通过广播区间Join替代逐行过滤，全程仅需对时序大表做1次扫描，计算效率可提升2~3个数量级。

from awsglue.dynamicframe import DynamicFrame
from pyspark.sql import functions as F

# 1. 转换pandas规则表为Spark DataFrame，统一时间字段类型为timestamp（避免字符串比较的性能问题和逻辑错误）
cond_df = spark.createDataFrame(pandas_df)
cond_df = cond_df.withColumn("start_ts", F.to_timestamp("start_ts")) \
                 .withColumn("end_ts", F.to_timestamp("end_ts"))

# 2. 转换原始时序DynamicFrame为Spark DataFrame，统一时间字段类型
ts_df = ts_dynamicframe.toDF()
ts_df = ts_df.withColumn("ts", F.to_timestamp("ts"))

# 3. 广播小体量规则表，执行区间Join：匹配id + ts落在对应时间区间内
# 规则表仅数千行，广播后无大表shuffle开销
joined_result = F.broadcast(cond_df).join(
    ts_df,
    on = (
        (ts_df["id"] == cond_df["id"]) &
        (ts_df["ts"] >= cond_df["start_ts"]) &
        (ts_df["ts"] <= cond_df["end_ts"])
    ),
    how = "inner"
)

# 4. 去重后保留原始时序表字段，转回DynamicFrame供后续Glue作业使用
result_df = joined_result.select(ts_df["id"], ts_df["val"], ts_df["ts"]).dropDuplicates()
result_dyf = DynamicFrame.fromDF(result_df, glueContext, "result_dyf")

# 触发计算
print("Count: ", result_dyf.count())

可选优化（id基数较大时适用） #

如果时序表id基数较高，可提前对两张表按id字段重分区，将相同id的数据分发到同一个Executor，进一步减少跨节点数据传输开销：

ts_df = ts_df.repartition("id")
cond_df = cond_df.repartition("id")

• 禁止用逐行遍历、Python UDF的方式操作分布式大表，这类操作会完全丢失分布式计算的并行优势
• 千行/万行级别的小表 join 百万/亿级大表时，必须加broadcast()提示强制广播小表，避免大表全局shuffle
• 所有时间字段必须提前转换为timestamp类型再做比较，直接用字符串比较既存在性能问题，也容易因格式不一致出现逻辑错误
• 禁止通过循环union拼接多个DataFrame，该方式会让执行计划长度随数据集数量线性增长，Spark优化阶段就会卡住

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