PySpark子集DataFrame数据不一致问题及强制非延迟执行方法

问题原因 #

Spark的惰性执行机制意味着所有转换操作（如limit、sort）仅构建逻辑执行计划，不会立即计算数据。

你用df_sample = df.limit(10)创建的并不是一个固定的数据集，而是一个指向原df取10行的逻辑计划。第一次调用df_sample.show()时触发了物理执行，Spark从原df中取出10行展示，但并没有把这10行持久化下来。

当后续执行df_temp的show()时，相当于重新执行整个逻辑链：原df → limit(10) → sort → groupBy。而Spark的limit在未物化数据的情况下，每次执行可能从原df的不同分区或位置取10行（因为分布式数据的读取顺序不固定），所以会出现df_temp包含原df中其他行的情况。

至于再次执行df_sample.show()时显示最初的10行，这只是Spark作业调度偶然复用了之前的计算结果，并非必然行为。

解决方法 #

要强制Spark固定df_sample的数据，需要将其物化，也就是把逻辑计划转换成实际存储的固定数据集，常用两种方式：

1. 缓存数据（推荐） #

使用cache()或persist()将df_sample的数据缓存到内存或磁盘，后续操作直接读取缓存的固定数据：

# 创建样本后立即缓存
df_sample = df.limit(10).cache()
# 触发缓存计算（必须执行一次行动操作，比如show()）
df_sample.show()

# 后续操作都基于缓存的固定数据
df_temp = df_sample.sort(F.desc('timestamp')).groupBy('id').agg(F.collect_list('value').alias('newcol'))
df_temp.show()

2. 转成本地集合再转回DataFrame #

将样本数据拉到Driver端，再创建新的DataFrame，适合极小数据集（如limit(10)）：

# 先把limit的结果拉到本地，再创建新的DataFrame
sample_data = df.limit(10).collect()
df_sample = spark.createDataFrame(sample_data, df.schema)

# 后续操作基于固定的本地数据
df_temp = df_sample.sort(F.desc('timestamp')).groupBy('id').agg(F.collect_list('value').alias('newcol'))
df_temp.show()

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