我给你的建议是：在小数据集上建立第一个联合，然后广播数据集，第一个联合的结果，spark将把数据集部署在它的不同节点上，这将减少随机数。spark上的union经过了很好的优化，因此您需要做的是考虑以下问题：只从一开始选择您需要的列，避免在union之前执行任何类型的非成本有效的操作，如groupbykey…等等，因为spark在生成最终进程时会调用这些操作。我建议您避免使用hive，因为它使用map reduce策略，与spark sql相比，这是不值得的。您可以使用此函数示例只需更改键，如果可以，请使用scala它将直接与spark交互：

def map_To_cells(df1: DataFrame, df2: DataFrame): DataFrame = {
val df0= df2.withColumn("key0",F.col("key")).drop("key")
df1.as("main").join(
broadcast(df0),
df0("key0") <=> df("key")
).select( needed columns)
}

你可以在每年的表上使用扣环 primary_key 以及 year 将列放入完全相同数量的bucket中，以避免在计算最终联接时进行昂贵的交换。

- output: table_type_1_year_0
  input: raw_table_type_1_year_0
  hive_partitioning: none
  bucketing: BUCKET_COUNT by (PRIMARY_KEY, YEAR)
- output: table_type_1_year_1
  input: raw_table_type_1_year_1
  hive_partitioning: none
  bucketing: BUCKET_COUNT by (PRIMARY_KEY, YEAR)
...
- output: table_type_2_year_0
  input: raw_table_type_2_year_0
  hive_partitioning: none
  bucketing: BUCKET_COUNT by (PRIMARY_KEY, YEAR)
- output: table_type_2_year_1
  input: raw_table_type_2_year_1
  hive_partitioning: none
  bucketing: BUCKET_COUNT by (PRIMARY_KEY, YEAR)
...
- output: all_tables
  input:
    - table_type_1_year_0
    - table_type_1_year_1
...
    - table_type_2_year_0
    - table_type_2_year_1
...
  hive_partitioning: none
  bucketing: BUCKET_COUNT by (PRIMARY_KEY, YEAR)

注意：当您选择 BUCKET_COUNT 价值观，重要的是要明白它应该为最终的目标而优化 all_tables 输出，而不是中间表。这意味着您可能会得到对于中间表来说非常小的文件。这可能是无关紧要的相比，效率的提高 all_tables 输出，因为在连接所有东西时，不必计算大量的交换；你的桶将预先计算，你可以简单地 SortMergeJoin 在输入文件上。
对于一个如何编写转换并写出指定数量的bucket的显式示例，我在这里的答案可能很有用。