PySpark多次运行结果不一致：重复ID异常排查求助

问题描述 #

多表关联生成DataFrame后，进行嵌套结构的聚合处理，排查id3重复时发现：每次执行排查操作，结果中重复的id3值随机变化，但重复数量固定。根据代码设计逻辑，id3应唯一，且验证显示原始关联操作未产生重复数据。

复现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
import pyspark.sql.functions as f
from pyspark.sql.functions import lit

# Create a Spark session
spark = SparkSession.builder.appName("CreateDataFrame").getOrCreate()

# User input for number of rows
n_a = 10
n_a_c = 5
n_a_c_d = 3
n_a_c_e = 4

# Define the schema for the DataFrame
schema_a = StructType([StructField("id1", StringType(), True)])
schema_a_b = StructType(
    [
        StructField("id1", StringType(), True),
        StructField("id2", StringType(), True),
        StructField("extra", StringType(), True),
    ]
)
schema_a_c = StructType(
    [
        StructField("id1", StringType(), True),
        StructField("id3", StringType(), True),
    ]
)
schema_a_c_d = StructType(
    [
        StructField("id3", StringType(), True),
        StructField("id4", StringType(), True),
    ]
)
schema_a_c_e = StructType(
    [
        StructField("id3", StringType(), True),
        StructField("id5", StringType(), True),
    ]
)

# Create a list of rows with increasing integer values for "id1" and a constant value of "1" for "id2"
rows_a = [(str(i),) for i in range(1, n_a + 1)]
rows_a_integers = [str(i) for i in range(1, n_a + 1)]
rows_a_b = [(str(i), str(1), "A") for i in range(1, n_a + 1)]


def get_2d_list(ids_part_1: list, n_new_ids: int):
    rows = [
        [
            (str(i), str(i) + "_" + str(j))
            for i in ids_part_1
            for j in range(1, n_new_ids + 1)
        ]
    ]
    return [item for sublist in rows for item in sublist]


rows_a_c = get_2d_list(ids_part_1=rows_a_integers, n_new_ids=n_a_c)
rows_a_c_d = get_2d_list(ids_part_1=[i[1] for i in rows_a_c], n_new_ids=n_a_c_d)
rows_a_c_e = get_2d_list(ids_part_1=[i[1] for i in rows_a_c], n_new_ids=n_a_c_e)

# Create the DataFrame
df_a = spark.createDataFrame(rows_a, schema_a)
df_a_b = spark.createDataFrame(rows_a_b, schema_a_b)
df_a_c = spark.createDataFrame(rows_a_c, schema_a_c)
df_a_c_d = spark.createDataFrame(rows_a_c_d, schema_a_c_d)
df_a_c_e = spark.createDataFrame(rows_a_c_e, schema_a_c_e)

# Join everything
df_join = (
    df_a.join(df_a_b, on="id1")
    .join(df_a_c, on="id1")
    .join(df_a_c_d, on="id3")
    .join(df_a_c_e, on="id3")
)

# Nested structure
# show
df_nested = df_join.withColumn("id3", f.struct(f.col("id3")))

for i, index in enumerate([(5, 3), (4, 3), (3, None)]):
    remaining_columns = list(set(df_nested.columns).difference(set([f"id{index[0]}"])))
    df_nested = (
        df_nested.groupby(*remaining_columns)
        .agg(f.collect_list(f.col(f"id{index[0]}")).alias(f"id{index[0]}_tmp"))
        .drop(f"id{index[0]}")
        .withColumnRenamed(
            f"id{index[0]}_tmp",
            f"id{index[0]}",
        )
    )

    if index[1]:
        df_nested = df_nested.withColumn(
            f"id{index[1]}",
            f.struct(
                f.col(f"id{index[1]}.*"),
                f.col(f"id{index[0]}"),
            ).alias(f"id{index[1]}"),
        ).drop(f"id{index[0]}")

# Investigate for duplicates in id3 (should be unique)
df_test = df_nested.select("id2", "extra", f.explode(f.col("id3")["id3"]).alias("id3"))

for i in range(5):
    df_test.groupby("id3").count().filter(f.col("count") > 1).show()

现象 #

每次执行循环输出时，会随机出现不同的id3被标记为重复，例如：
某次输出：

+---+-----+
|id3|count|
+---+-----+
|6_4|    2|
+---+-----+

另一次输出：

+---+-----+
|id3|count|
+---+-----+
|9_3|    2|
+---+-----+

排查验证 #

执行df_join.groupby("id3", "id4", "id5").count().filter(f.col("count") > 1).show()返回空结果，证明多表关联操作本身未产生重复数据，问题出在后续的嵌套聚合步骤。

问题根源 #

问题出在循环中生成remaining_columns的方式：使用set处理列名，而set是无序的，导致每次运行时groupby的列顺序随机变化。在Spark 3.3.0（Databricks Runtime 11.3 LTS）中，当struct列参与分组且列顺序随机时，会触发内部分组逻辑异常，导致错误地将重复的id3收集到同一个分组中，最终explode后出现重复。

解决方案 #

将生成remaining_columns的代码改为保持列的原始顺序，避免使用无序的set：

# 替换原有的set处理方式
remaining_columns = [col for col in df_nested.columns if col != f"id{index[0]}"]

修改后每次运行时groupby的列顺序固定，分组逻辑一致，不会再出现随机重复的id3。

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