高效的pyspark join

Question

我已经阅读了很多关于如何在 pyspark 中进行高效连接的文章。我发现的实现高效连接的方法基本上是：

• 如果可以，请使用广播加入。 （我通常做不到，因为数据框太大）
• 考虑使用一个非常大的集群。 （我宁愿不要因为 $$$）。
• 使用相同的分区器。

最后一个是我宁愿尝试的，但我在 pyspark 中找不到方法。我试过了：

df.repartition(numberOfPartitions,['parition_col1','partition_col2'])

但这无济于事，直到我停止它仍然需要很长时间，因为 spark get 卡在最后几个工作中。

那么，我如何在 pyspark 中使用相同的分区器并加快我的连接速度，甚至摆脱需要永远进行的洗牌？我需要使用哪个代码？

PD：我查看了其他文章，甚至在 stackoverflow 上，但我仍然看不到代码。

Answer 1

如果适合您的要求，您也可以使用两遍方法。首先，重新分区数据并使用分区表 (dataframe.write.partitionBy()) 进行持久化。然后，在一个循环中连续连接子分区，“附加”到同一个最终结果表。 Sim很好地解释了这一点。请看下面的链接

two pass approach to join big dataframes in pyspark

根据上面解释的案例，我能够在一个循环中串行连接子分区，然后将连接的数据持久化到配置单元表中。

这里是代码。

from pyspark.sql.functions import *
emp_df_1.withColumn("par_id",col('emp_id')%5).repartition(5, 'par_id').write.format('orc').partitionBy("par_id").saveAsTable("UDB.temptable_1")
emp_df_2.withColumn("par_id",col('emp_id')%5).repartition(5, 'par_id').write.format('orc').partitionBy("par_id").saveAsTable("UDB.temptable_2")

因此，如果您要加入整数 emp_id，您可以按 ID 模数进行分区，这样您就可以在 spark 分区之间重新分配负载，并且具有相似键的记录将被分组在一起并驻留在同一个分区上. 然后，您可以读取并循环遍历每个子分区数据，并将两个数据帧连接起来并将它们持久化。

counter =0;
paritioncount = 4;
while counter<=paritioncount:
    query1 ="SELECT * FROM UDB.temptable_1 where par_id={}".format(counter)
    query2 ="SELECT * FROM UDB.temptable_2 where par_id={}".format(counter)
    EMP_DF1 =spark.sql(query1)
    EMP_DF2 =spark.sql(query2)
    df1 = EMP_DF1.alias('df1')
    df2 = EMP_DF2.alias('df2')
    innerjoin_EMP = df1.join(df2, df1.emp_id == df2.emp_id,'inner').select('df1.*')
    innerjoin_EMP.show()
    innerjoin_EMP.write.format('orc').insertInto("UDB.temptable")
    counter = counter +1

我已经尝试过了，而且效果很好。这只是演示两遍方法的示例。您的连接条件可能会有所不同，分区的数量也取决于您的数据大小。

Answer 2

感谢@vikrantrana 的回答，如果需要，我会尝试。我之所以这么说是因为我发现问题不在于“大”连接，问题在于连接之前的计算量。想象一下这种情况：

我读取了一个表并将其存储在一个名为 df1 的数据框中。我读了另一张表，并将其存储在df2 中。然后，我执行了大量的计算并连接到两者，最终得到df1 和df2 之间的连接。这里的问题不是大小，问题是spark的执行计划很大，它无法维护内存中的所有中间表，所以它开始写入磁盘，花了很多时间。

对我有用的解决方案是在连接之前将df1 和df2 保存在磁盘中（我还保存了其他中间数据帧，它们是大型复杂计算的结果）。