PySpark DataFrame与toPandas()转换后数据不一致问题

问题重现 #

执行以下PySpark操作后，Spark中查询TEMPERATURE列最大值为38，但转换为Pandas DataFrame后，该列最大值变为659，数据结果不一致：

import pyspark.sql.functions as F
from pyspark.sql.functions import when, lit, monotonically_increasing_id

df = df.withColumn("ID", monotonically_increasing_id())

anomaly_ids = [15699, 48526, 81592]

# 将异常ID对应行的非ID、TIMESTAMP列值设为None
for col_name in [col for col in df.columns if col not in ["ID", "TIMESTAMP"]]:
    df = df.withColumn(col_name, when(df["ID"].isin(anomaly_ids), lit(None)).otherwise(df[col_name]))

Spark查询结果：

df.select(F.max(col("TEMPERATURE"))).show()

+--------------------------+
| max(TEMPERATURE) |
+--------------------------+
| 38|
+--------------------------+

转Pandas后查询结果：

dfx = df.select("*").toPandas()
dfx["TEMPERATURE"].max()

659.0

核心原因 #

问题根源在于monotonically_increasing_id()的非确定性，结合Spark的惰性求值机制：

• monotonically_increasing_id()生成的ID依赖数据分区，每次触发计算（如show()、toPandas()）时，若分区逻辑变化，生成的ID会完全不同。
• 第一次调用show()触发计算时，生成的ID匹配了anomaly_ids，对应行的TEMPERATURE被设为None，因此最大值为38；但调用toPandas()时，会重新触发整个DataFlow的计算，此时生成的ID与之前不一致，原本的异常行未被标记处理，导致Pandas中保留了原始最大值659。

解决方案 #

需要将生成ID后的DataFrame物化，固定ID值，避免重复计算时ID变化：

方案1：缓存DataFrame #

在生成ID后调用cache()持久化数据，强制Spark固定ID值：

df = df.withColumn("ID", monotonically_increasing_id())
# 缓存DataFrame
df.cache()
# 触发行动操作完成缓存
df.count()

anomaly_ids = [15699, 48526, 81592]

for col_name in [col for col in df.columns if col not in ["ID", "TIMESTAMP"]]:
    df = df.withColumn(col_name, when(df["ID"].isin(anomaly_ids), lit(None)).otherwise(df[col_name]))

方案2：写入临时视图 #

将生成ID后的DataFrame写入临时视图，后续操作基于视图进行：

df = df.withColumn("ID", monotonically_increasing_id())
# 创建临时视图固定数据
df.createOrReplaceTempView("temp_df")
# 从视图重新读取数据
df = spark.sql("SELECT * FROM temp_df")

anomaly_ids = [15699, 48526, 81592]

for col_name in [col for col in df.columns if col not in ["ID", "TIMESTAMP"]]:
    df = df.withColumn(col_name, when(df["ID"].isin(anomaly_ids), lit(None)).otherwise(df[col_name]))

方案3：使用确定性ID生成方式 #

替换monotonically_increasing_id()为基于排序字段的row_number()，保证ID生成结果固定：

from pyspark.sql.window import Window

# 基于TIMESTAMP排序生成唯一ID，结果可重复
window = Window.orderBy("TIMESTAMP")
df = df.withColumn("ID", F.row_number().over(window))

anomaly_ids = [15699, 48526, 81592]

for col_name in [col for col in df.columns if col not in ["ID", "TIMESTAMP"]]:
    df = df.withColumn(col_name, when(df["ID"].isin(anomaly_ids), lit(None)).otherwise(df[col_name]))

验证 #

执行上述任一方案后，再次分别在Spark和Pandas中查询TEMPERATURE列最大值，结果会保持一致。

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