如何在 Spark SQL 中完成我的高度倾斜连接?

van*_*ser 2 apache-spark apache-spark-sql

我的加入执行如下:

SELECT
  left.*, 
  right.*
FROM `/foo/bar/baz` AS left
JOIN `/foo2/bar2/baz2` AS right
ON left.something = right.something
Run Code Online (Sandbox Code Playgroud)

数据集: /foo/bar/baz

+-----------+-------+
| something | val_1 |
+-----------+-------+
| a         |     1 |
| a         |     2 |
| a         |     3 |
| a         |     4 |
| a         |     5 |
| a         |     6 |
| a         |   ... |
| a         |   10K |
| b         |     1 |
| b         |     2 |
| b         |     3 |
+-----------+-------+
Run Code Online (Sandbox Code Playgroud)

数据集: /foo2/bar2/baz2

+-----------+-------+
| something | val_2 |
+-----------+-------+
| a         |     1 |
| a         |     2 |
| b         |     1 |
| b         |     2 |
| b         |     3 |
+-----------+-------+
Run Code Online (Sandbox Code Playgroud)

我的执行程序出现 OOM 错误,我不想不必要地向执行程序抛出更多内存。如何在不消耗额外资源的情况下确保此连接成功执行?

van*_*ser 5

盐渍联接

使此连接成功执行的一种策略是执行所谓的对连接进行加盐处理。

盐渍连接在 Spark 中通过将每个键具有多个条目的表拆分为较小的部分,同时将较小的表分解为相同数量的副本来运行。这导致与普通连接相同大小的输出,但较大表的任务大小较小,因此降低了 OOM 错误的风险。通过将一列从 0 到 N 的随机数添加到左表并制作右表的 N 个副本,可以对连接进行加盐。如果您将新的随机列添加到连接中,您会将最大的桶减少到其先前大小的 1/N。

秘密在于 EXPLODE 函数。EXPLODE 是一个交叉产品:

SELECT
  left.*, 
  right.*
FROM
  (
    SELECT 
      *, 
      FLOOR(RAND() * 8) AS salt 
      FROM `/foo/bar/baz`
  ) AS left
JOIN
  (
    SELECT 
      *, 
      EXPLODE(ARRAY(0,1,2,3,4,5,6,7)) AS salt 
      FROM `/foo2/bar2/baz2`
  ) AS right
ON 
left.something = right.something 
AND left.salt = right.salt
Run Code Online (Sandbox Code Playgroud)

调音

  • 你如何选择爆炸的因素?受过教育的猜测,主要是。2 的幂是找到合适的范围的好方法:8、16、32。
  • 类似的方法是在未加盐作业运行时查看每个执行程序的行数。

需要注意什么

  • 确保在对连接进行加盐时不会出现一一错误。这将使您丢失一小部分记录。
  • CEIL(RAND() * N)为您FLOOR(RAND() * N)提供 0 和 N 之间的整数。为您提供 0 和 N — 1 之间的数字。确保在加盐连接中分解正确的一组数字!

盐的开销

  • 对连接进行加盐并不一定会使您的构建速度更快。它只是给了它更好的成功机会。
  • 如果您不必要地对连接加盐,您实际上可能会开始看到性能下降。