用于有效连接Spark数据帧/数据集的分区数据

Question

我需要根据一些共享的键列将许多数据帧连接在一起。对于键值RDD，可以指定一个分区程序，以便具有相同键的数据点被洗牌到相同的执行器，因此连接更有效（如果在连接之前有与洗牌相关的操作）。可以在火花数据帧或数据集上做同样的事情吗？

Answer 1

可以使用重新分区方法使用DataFrame/DataSet API。使用此方法，您可以指定一个或多个列用于数据分区，例如。

val df2 = df.repartition($"colA", $"colB")

也可以同时在同一命令中指定所需分区的数量，

val df2 = df.repartition(10, $"colA", $"colB")

注意：这并不能保证数据帧的分区位于同一个节点上，只是分区以相同的方式完成。

Answer 2

如果知道要多次加入数据帧，可以在加载数据帧后对其进行重新分区

val users = spark.read.load("/path/to/users").repartition('userId)

val joined1 = users.join(addresses, "userId")
joined1.show() // <-- 1st shuffle for repartition

val joined2 = users.join(salary, "userId")
joined2.show() // <-- skips shuffle for users since it's already been repartitioned

因此，它将对数据进行一次洗牌，然后在加入后续时间时重用洗牌文件。

然而，如果你知道你会在某些键上反复洗牌数据，最好的办法是将数据保存为带扣的表。这将把已经预哈希分区的数据写出来，因此当您读入并加入表时，可以避免混乱。你可以这样做：

// you need to pick a number of buckets that makes sense for your data
users.bucketBy(50, "userId").saveAsTable("users")
addresses.bucketBy(50, "userId").saveAsTable("addresses")

val users = spark.read.table("users")
val addresses = spark.read.table("addresses")

val joined = users.join(addresses, "userId")
joined.show() // <-- no shuffle since tables are co-partitioned

为了避免洗牌，桌子必须使用相同的扣板（例如相同数量的桶和桶柱上的连接）。