apache-spark - PySpark:加入后不一致的 count() 结果

Question

我对以下问题感到非常困惑:

当我加入 2 个数据框并返回行数时，每次尝试的计数都略有不同。以下是详细信息:

我想加入数据框:“df_user_ids”和“df_conversions”:

df_user_ids.show()
>>>
+--------------------+
|             user_id|
+--------------------+
|AMsySZY-cqcufnXst...|
|AMsySZY1Oo75A6vKU...|
|AMsySZY4nbqZiuEMR...|
|AMsySZY5RSfgj6Xvi...|
|AMsySZY5geAmTx0er...|
|AMsySZY6Gskv_kEAv...|
|AMsySZY6MIOyPWM4U...|
|AMsySZYCEZYS00UB9...| 

---

df_conversions.show()
>>>
+--------------------+----------------------+---------+
|             user_id|time_activity_observed|converted|
+--------------------+----------------------+---------+
|CAESEAl1YPOZpaWVx...|   2018-03-23 12:15:37|        1|
|CAESEAuvSBzmfc_f3...|   2018-03-23 21:58:25|        1|
|CAESEBXWsSYm4ntvR...|   2018-03-30 12:16:53|        1|
|CAESEC-5uPwWGFdnv...|   2018-03-23 08:52:48|        1|
|CAESEDB3Z-NNvz7zL...|   2018-03-24 21:37:05|        1|
|CAESEDu7S7rGTVlj2...|   2018-04-01 17:00:12|        1|
|CAESEE4s6g1-JlUEt...|   2018-03-23 19:32:23|        1|
|CAESEELlJt0mE2xjn...|   2018-03-24 18:26:15|        1|

两个数据框都有名为“user_id”的键列，两者都是使用带有固定种子的“.sampleBy()”创建的:

.sampleBy("converted", fractions={0: 0.035, 1: 1}, seed=0)    

在加入数据框之前，我将它们保存到磁盘:

df_user_ids.persist(StorageLevel.DISK_ONLY)
df_conversions.persist(StorageLevel.DISK_ONLY) 

然后我验证两个数据帧的行数是否一致:

df_user_ids.count()
>>> 584309

df_user_ids.count()
>>> 584309

df_conversions.count()
>>> 5830

df_conversions.count()
>>> 5830

并检查两个数据框的键列不包含重复项:

df_user_ids.count()
>>> 584309

df_user_ids.select('user_id').distinct().count()
>>> 584309

df_conversions.count()
>>> 5830

df_conversions.select('user_id').distinct().count()
>>> 5830

然后当我加入它们时我得到不一致的行数!

df_user_ids.join(df_conversions, ["user_id"], "left").count()
>>> 584314

df_user_ids.join(df_conversions, ["user_id"], "left").count()
>>> 584317

df_user_ids.join(df_conversions, ["user_id"], "left").count()
>>> 584304

这怎么可能？？

有时这个加入计数高于“df_user_ids.count()”，有时低于。我在 EMR 集群上的 AWS EMR 中使用 Zeppelin notebook 来运行此代码。

我已经尝试过以下链接中的建议:

• ".persist(StorageLevel.DISK_ONLY)"没有帮助。
• 我不使用 monotonically_increasing_id。

spark inconsistency when running count command

Answer 1

通过查看您在 DataFrames 上执行的一系列操作，我认为问题出在 Join 上。加入操作结果随机播放，其中每个节点都与其他每个节点通信，并且它们根据哪个节点具有特定键或一组键(您正在加入)共享数据。在执行器之间共享数据时，如果执行器没有将数据帧持久保存在磁盘上，它将重新计算 DAG，并且 sampleBy 不能保证返回数据帧中相同的行数。