Spark >2 - 连接操作期间的自定义分区键

Question

我想知道我们是否可以强制 Spark 在两个数据帧的连接操作期间使用自定义分区键。

例如，让我们考虑

df1: DataFrame - [groupid, other_column_a]
df2: DataFrame - [groupid, other_column_b]

如果我跑步

df_join = df1.join(df2, "groupid")

Spark 会将“groupid”设置为分区键，并在每个分区上执行连接。问题是，如果分区太大，这可能会耗尽机器上的内存。

但是，理论上似乎可以使用(groupid, other_column_a) 作为分区键执行连接（以减小每个分区的大小）。

Spark 可以做到吗？我试着做 df1.repartition("group_id","other_column_a") 前期，但这被连接覆盖（我用 df_join.explain() 检查它）。我在网上找不到任何解释如何执行此操作的资源。

谢谢！

Answer 1

如果您要加入我想要的某个整数列，您可以对您的数据框进行分区，我将取模一些数字.. 即您想要多少个分区.. 这样共享一些常见哈希值的 id 将是组合在一个分区中..然后您可以通过分解多个连接来执行连接..在循环中连续连接每个分区..我已经详细解释了这个案例..Efficient pyspark join

Answer 2

但是，理论上似乎可以使用 say (groupid, other_column_a) 执行连接

这是不正确的。要执行连接，Spark 必须将带有 groupid 的所有记录移动到单个分区，因此在以下情况下使用 (groupid, other_column_a) 是可能的：

• (groupid, other_column_a) where 连接键。
• other_column_a 和 group_id 之间存在函数关系。

显然不满足第一个条件，因为您仅通过groupid 加入，第二个无法解决问题，因为分布相同或更差。

对于倾斜连接还有其他可能的解决方案，例如单独处理倾斜组或迭代广播连接（请参阅Spark final task takes 100x times longer than first 199, how to improve 中的答案和 cmets）。