apache-spark - 在通过唯一键连接 DataFrame 时如何避免洗牌？

Question

我有两个 DataFrame A 和 B:

• A 包含列 (id, info1, info2)，大约有 2 亿行
• B 只有 id 列有 100 万行

id 列在两个 DataFrame 中都是唯一的。

我想要一个新的 DataFrame，它可以过滤 A 以仅包含 B 中的值。

如果 B 非常小，我知道我会做类似的事情

A.filter($("id") isin B("id"))

但是B仍然很大，所以并不是所有的都可以作为广播变量。

我知道我可以使用

A.join(B, Seq("id"))

但这不会利用独特性，而且恐怕会导致不必要的洗牌。

完成该任务的最佳方法是什么？

Answer 1

如果您尚未在 Dataframe A 上应用任何分区器，这可能会帮助您理解 Join 和 Shuffle 概念。

没有分区器:

A.join(B, Seq("id"))

默认情况下，此操作将对两个数据帧的所有键进行哈希处理，通过网络将具有相同键哈希的元素发送到同一台机器，然后在该机器上将具有相同键的元素连接在一起。在这里，您必须注意到两个数据帧都在网络中随机播放。

使用 HashPartitioner:在构建Dataframe时调用partitionBy()，Spark现在将知道它是哈希分区的，并且对其调用join()将利用此信息。特别是，当我们调用 A.join(B, Seq("id")) 时，Spark 将仅对 B RDD 进行混洗。由于 B 的数据少于 A，因此无需在 B 上应用分区器

例如:

 val A = sc.sequenceFile[id, info1, info2]("hdfs://...")
     .partitionBy(new HashPartitioner(100)) // Create 100 partitions
     .persist()
 A.join(B, Seq("id"))

引用资料来自《Learning Spark》一书。