火花加入*无*随机播放

Question

我正在尝试优化我的火花应用工作。

我试图理解这个问题的要点:如何在唯一键上连接数据帧时避免混乱？

>

我已经确保必须发生加入操作的键分布在同一分区中（使用我的自定义分区程序）。

我也不能做广播加入，因为我的数据可能会根据情况变大。

在上面提到的问题的答案中，重新分区只优化了连接，但我需要的是无需切换即可连接。在分区内的键的帮助下，我对连接操作很满意。

有可能吗？如果不存在类似的功能，我想实现像joinperpartition这样的东西。

Answer 1

只是对之前好答案的补充。如果在整个pyspark应用程序中多次连接一个大的数据帧，那么将该表保存为分桶表，并在pyspark中作为数据帧读回。这样，您可以避免在连接过程中多次洗牌，因为数据已经被预先洗牌和排序。

因此，当Spark在两个大型数据帧上选择排序合并连接时，它会在连接操作中跳过排序和洗牌阶段。(可以边看wholecodegen边在spark UI确认)

df_data_1.coalesce(1).write.format('orc').bucketBy(20, 'joincolumn').sortBy("sortcolumn").mode("overwrite").saveAsTable('bucketed_table1')

df_data_2.coalesce(1).write.format('orc').bucketBy(20, 'joincolumn').sortBy("sortcolumn").mode("overwrite").saveAsTable('bucketed_table2')

df_bucket_table_1 = spark.table("bucketed_table1");
df_bucket_table_2 = spark.table("bucketed_table2");

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
spark.conf.set("spark.sql.join.preferSortMergeJoin","true")

 #creating alias for the dataframes:
from pyspark.sql.functions import *

df1 = df_bucket_table_1.alias('df1')
df2 = df_bucket_table_2.alias('df2')


DfInnerJoin = df1.join(df2, df1.joincolumn == df2.joincolumn,'inner').select('df1.*')

上述联接不会有随机排列，但仅当您必须在整个应用程序中多次联接同一数据帧时，这才有用。

Answer 2

重新分区只能优化连接，但我需要的是无需洗牌的连接

这不是真的。重新分区不仅“优化”了连接。Repartition将< code>Partitioner绑定到您的RDD，这是映射端连接的关键组件。

我已经确保必须进行连接操作的键分布在同一个分区中

Spark必须知道这一点。使用适当的 API 构建数据帧，以便它们具有相同的分区程序，Spark 将负责其余的工作。