为什么PySpark dropDuplicates和Join会产生奇怪的结果

Question

在dropDuplicates和join数据集之后，PySpark给出了一些奇怪的结果。情况是有两个非常大的数据集：一个是人的ID和一些变量，第二个是他们的地区代码

第一个数据集：

ID|VAR1|VAR2|VAR3|VAR4|VAR5|
1|-----|----|---|---|----|
2|-----|----|---|---|----|
3|-----|----|---|---|----|
4|-----|----|---|---|----|

第二个数据集：

^{2}$

我得到的结果是：

file_1 = file_1.dropDuplicates(["ID"])
file_2 = file_2.dropDuplicate(["ID"])
file_2.filter(filter("ID == '1'").show()

ID|region_code|
1|7|

加入文件后，我希望：

merge_file = file_1.join(file_2, "ID", "left")

ID|VAR1|VAR2|VAR3|VAR4|VAR5|region_code|
1|-----|----|---|---|----|7|
2|-----|----|---|---|----|5|
3|-----|----|---|---|----|null|
4|-----|----|---|---|----|7|

但我得到了：

merge_file.filter("ID == '1'").show()

ID|VAR1|VAR2|VAR3|VAR4|VAR5|region_code|
1|-----|----|---|---|----|9|

我对这些奇怪的结果很好奇。在

Answer 1

pyspark是一个懒惰的解释器。您的代码只在调用操作时执行（即show()，count()等）。在代码示例中，您正在创建file_2。与其把file_2看作是一个存在于内存中的对象，file_2实际上只是一组指令，告诉pyspark引擎的处理步骤。当您调用file_2.filter(filter("ID == '1'").show()时，将执行这些指令（包括dropDuplicates()）以生成输出。在

当您创建merge_file时，您引用了file_2的处理步骤，其计算方法可能与前面的示例不同。不能保证dropDuplicates()每次都会删除相同的行，因为pyspark的延迟执行。这就是为什么在两个数据帧之间得到不同的结果。在

这也是为什么我建议您考虑一下您的数据以及您想用merge完成什么，因为如果{}和{}之间的关系很重要，dropDuplicates()不是一个可靠的方法。在