如何解决PySpark非确定性函数导致Redshift数仓ETL结果不一致问题

问题根源 #

你遇到的ETL运行结果不稳定问题和UDF非确定性关联不大，核心原因有2个：

• groupBy操作不会保证输出行的固定顺序，你直接在聚合后调用monotonically_increasing_id()生成维度表主键，每次运行时同个维度组合对应的行顺序可能发生变化，导致生成的主键ID不一致
• coalesce(1)仅能将分区合并为单分区，不会固定行的排序规则，无法解决顺序不稳定的问题

解决方案 #

1. 强制维度表生成前按固定规则排序 #

在生成维度主键前增加orderBy逻辑，按维度字段稳定排序，确保每次运行同个维度组合对应行的顺序固定，生成的ID保持一致：

physical_attribute_df = (
    cat_df.select(
        F.coalesce(
            animal_color_mapper[cat_df.fur_color],
            F.lit("NOT SPECIFIED")
        ).cast('string').alias('color'),
        F.coalesce(
            animal_color_mapper2(cat_df.fur_color),
            F.lit("NOT SPECIFIED")
        ).cast('string').alias('color2'),
        cat_df.zoo_id.alias('_zoo_id'),
    ).groupBy(
        'color',
        'color2',
    ).agg(
        F.collect_list('_zoo_id').alias('_zoo_ids')
    ).coalesce(1)
    # 新增稳定排序逻辑，维度字段可根据实际业务调整
    .orderBy('color', 'color2')
    .withColumn(
        'id', F.monotonically_increasing_id()
    ).withColumn(
        'created_date', F.current_timestamp()
    ).withColumn(
        'last_updated', F.current_timestamp()
    )
)

2. （可选）标记UDF为确定性函数 #

PySpark 2.3及以上版本支持将UDF标记为确定性函数，避免Spark因误判非确定性导致的重复计算问题：

@F.udf(returnType='string')
def animal_color_mapper2(x):
    return animal_color_map.get(x, "NOT SPECIFIED")

# 标记为确定性函数
animal_color_mapper2 = animal_color_mapper2.asDeterministic()

3. 修复代码拼写错误 #

你的代码中存在变量名拼写不一致问题，未修正的话会导致关联空值：

• join操作中用到的_physical_attributes_ids_df实际定义的变量名为_physical_attributes_df
• 关联后取的字段physical_attributes_id实际生成的字段名为physical_attribute_id

4. （推荐）优化映射逻辑，优先使用内置函数 #

create_map方案本身比UDF性能更优，可简化未匹配值的处理逻辑：

# 替换原coalesce写法，更简洁稳定
animal_color_mapper[cat_df.fur_color].cast('string').na.fill("NOT SPECIFIED").alias('color')

内容的提问来源于stack exchange，提问作者Justin B.