星星之火- sortWithInPartitions超过排序

Question

下面是表示员工in_date和out_date的示例数据集。我必须获得所有员工中的最后一个in_time。

星星之火运行在4节点独立集群上。

初始数据集：

雇员-日期

1111111     2017-04-20  2017-09-14 
1111111     2017-11-02  null 
2222222     2017-09-26  2017-09-26 
2222222     2017-11-28  null 
3333333     2016-01-07  2016-01-20 
3333333     2017-10-25  null 

df.sort(col(in_date).desc())后的数据集

雇员--日期

1111111   2017-11-02   null 
1111111   2017-04-20   2017-09-14 
2222222   2017-09-26   2017-09-26 
2222222   2017-11-28   null 
3333333   2017-10-25   null 
3333333   2016-01-07   2016-01-20 

df.dropDup(EmployeeID):  

输出：

雇员-日期

1111111    2017-11-02    null 
2222222    2017-09-26    2017-09-26 
3333333    2016-01-07    2016-01-20 

预期数据集:

雇员-日期

1111111    2017-11-02   null 
2222222    2017-11-28   null 
3333333    2017-10-25   null 

但是，当我使用sortWithInPartitions对初始数据集进行排序并销毁时，我得到了预期的数据集。我在这里漏掉了什么东西吗？任何帮助都是非常感谢的。

附加信息:当df.sort在本地模式下使用Spark执行时，实现了上述预期输出。

我没有做过任何分区，重新划分。初始数据集来自底层的Cassandra数据库。

Answer 1

TL;DR，除非明确保证，否则您永远不应该假设Spark中的操作将以任何特定的顺序执行，特别是在使用Spark时。

你现在缺的是洗牌。dropDuplicates实现相当于：

df.groupBy(idCols).agg(first(c) for c in nonIdCols)

将以下列方式执行：

• 部分(“映射端”)聚合。
• 随机播放。
• 最后(“还原侧”)聚合。

中间洗牌引入了非确定性，并且不能保证最终的聚合将以任何特定的顺序应用。

当在本地模式下执行df.sort时，实现了上述预期输出。

local模式相当简单。您不应该使用它来得出关于星火内部在完全分布式模式下的行为的结论。

当我使用sortWithInPartitions对初始数据集进行排序并销毁时，我得到了预期的数据集。

如果数据以前是由EmployeeID分区的，这将是有意义的。在这种情况下，星火将不需要额外的洗牌。

根据描述，您应该使用How to select the first row of each group?中显示的解决方案之一。