PySpark基于另一DataFrame创建新列及SPARK-5063错误解决

核心思路 #

避开易触发序列化错误（SPARK-5063）的UDF，改用PySpark内置函数、窗口函数和DataFrame原生操作实现规则，既解决序列化问题，又保证计算性能。

具体实现步骤 #

1. 预处理运费表（宽表转窄表） #

原dhl_price是宽表（A/B/C/D列为不同类别运费），先转为窄表，方便后续按类别关联：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 将宽表转为窄表，统一Type和运费字段
dhl_long = dhl_price.select(
    F.col("Weight"),
    F.explode(
        F.array(
            F.struct(F.lit("A").alias("Type"), F.col("A").alias("delivery_fee")),
            F.struct(F.lit("B").alias("Type"), F.col("B").alias("delivery_fee")),
            F.struct(F.lit("C").alias("Type"), F.col("C").alias("delivery_fee")),
            F.struct(F.lit("D").alias("Type"), F.col("D").alias("delivery_fee"))
        )
    ).alias("type_fee")
).select(
    F.col("Weight"),
    F.col("type_fee.Type"),
    F.col("type_fee.delivery_fee")
)

2. 为请求表添加唯一标识（可选） #

如果requests没有唯一主键，先添加临时ID，确保后续分组计算能定位到单条请求：

requests_with_id = requests.withColumn("req_id", F.monotonically_increasing_id())

3. 关联并筛选符合条件的运费记录 #

将请求表和预处理后的运费表按Type关联，筛选所有重量≥商品重量的记录，再通过窗口函数取最小重量对应的运费：

# 按Type关联请求表和运费表，筛选重量符合条件的记录
joined = requests_with_id.join(
    dhl_long,
    on="Type",
    how="left"
).filter(
    F.col("Weight") >= F.col("product_weight")
)

# 按请求ID和Type分组，按Weight升序排序后取第一条（最小重量的运费）
win = Window.partitionBy("req_id", "Type").orderBy("Weight")
result = joined.withColumn(
    "row_rank",
    F.row_number().over(win)
).filter(
    F.col("row_rank") == 1
).select(
    # 保留原请求表字段+新增的delivery_fee
    "req_id", "product_weight", "Type", "delivery_fee"
).drop("req_id")  # 不需要临时ID可删除

规则适配说明 #

• 当product_weight ≤30时：若dhl_price存在对应重量的记录，筛选后第一条就是匹配价格；若不存在，自动取≥该重量的最小重量价格，符合规则要求。
• 当product_weight >30时：直接取≥该重量的最小重量对应价格，完全符合规则。

方案优势 #

• 全程使用PySpark原生API，避免UDF带来的序列化问题（SPARK-5063）。
• 窗口函数和原生关联操作的性能远优于UDF，适合大数据量场景。

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