scala - Spark distinct 的实现

Question

我是 Spark 和 Scala 的新手。我正在阅读 distinct() 的 Spark 函数。但我找不到任何合适的细节。我有一些疑惑，我无法解决并写下来。

1. distinct() 在 Spark 中是如何实现的？

   我不太了解 Spark 源代码，无法识别整个流程。当我检查执行计划时，我只能看到一个 ShuffleRDD

2. distinct 的时间复杂度是多少？

   我还通过谷歌搜索发现它也以某种方式使用了散列和排序。

   所以，我想它是否使用与借助 Hashset 从数组中获取唯一元素相同的原理。如果是一个系统，我猜时间复杂度是 O(nlogn) 。

   但是分布在很多partitions中打乱，时间复杂度的顺序是多少？

3. 有没有办法在特定情况下避免洗牌？

   如果我确保根据我的用例正确分区我的数据，我可以避免洗牌吗？

   即例如，假设在具有唯一行的数据框中分解 ArrayType 列会创建新行，其他列被复制。 我将选择其他列。 通过这种方式，我确保每个分区的重复项都是唯一的。 因为我知道每个分区的重复项都是唯一的， 我可以避免随机播放，只需敏锐地删除该分区中的重复项

我也找到了这个Does spark's distinct() function shuffle only the distinct tuples from each partition .

感谢您的帮助。如果我有任何错误，请纠正我。

Answer 1

How distinct() is implemented in Spark ?

通过应用具有 None 值的虚拟聚合。大致

rdd.map((_, None)).reduceByKey((a, b) => a)

What is the Time Complexity of distinct ?

鉴于整个过程的复杂性，很难估计。它至少是 O(N log N)，因为 shuffle 需要排序，但考虑到构建额外的核心数据结构(包括关联数组)所需的多个其他操作，序列化/反序列化数据可能更高，并且在实践中由 IO 主导操作，而不是纯粹的算法复杂性。

Is there a way to avoid shuffling in particular cases ?

是的，如果保证将潜在的重复项放置在同一分区上。

您可以使用 mapPartitions 对数据进行去重复，尤其是当数据已排序或以其他方式保证在孤立的邻域中具有重复项时。如果没有这个，你可能会受到内存要求的限制，除非你接受概率过滤器(如 Bloom 过滤器)的近似结果。

一般来说虽然不可能，但这样的操作将是非本地的。