Spark数据集/数据帧连接空倾斜键

Question

使用Spark Dataset/DataFrame联接时，我面临长时间运行且OOM作业失败的问题。

以下是输入：

• ~10个不同大小的数据集，大部分是巨大的（

经过一番分析，我发现作业失败和缓慢的原因是null歪斜键：当左侧有数百万条记录时，用连接键null。

我用了一些蛮力的方法来解决这个问题，这里我想和大家分享一下。

如果您有更好的或任何内置的解决方案（针对常规Apache Spark），请与他人分享。

Answer 1

以下是我想到的解决方案：

  /**
    * Expression that produce negative random between -1 and -`lowestValue`(inclusively).
    *
    * @example
    *          {{{
    *             spark
    *                  .range(1, 100)
    *                  .withColumn("negative", negativeRandomWithin(3))
    *                  .select("negative")
    *                  .distinct()
    *                  .show(false)
    *          }}}
    *          +--------+
    *          |negative|
    *          +--------+
    *          |-2      |
    *          |-3      |
    *          |-1      |
    *          +--------+
    */
  private[transformation] def negativeRandomWithin(lowestValue: Long): Column = {
    negate(positiveRandomWithin(lowestValue)) - 1
  }

  /**
    * Expression that produce positive random between 0 and `highestValue`(exclusively).
    *
    * @example
    *          {{{
    *             spark
    *                  .range(1, 100)
    *                  .withColumn("positive", positiveRandomWithin(3))
    *                  .select("positive")
    *                  .distinct()
    *                  .show(false)
    *          }}}
    *          +--------+
    *          |positive|
    *          +--------+
    *          |0       |
    *          |1       |
    *          |2       |
    *          +--------+
    */
  private[transformation] def positiveRandomWithin(highestValue: Long) = {
    pmod((rand * highestValue).cast(LongType), lit(highestValue))
  }

  implicit class SkewedDataFrameExt(val underlying: DataFrame) extends AnyVal {

    /**
      * Particular optimized version of left outer join where left side of join has skewed `null` field.
      *
      * @note
      *       It works only for single column join which is applicable for `isNotNull`.
      *
      * Optimization algorithm:
      *   1. replace left dataset `null` values with negative number within range between -1 and - `nullNumBuckets`(10000 by default)
      *   2. use appended column, with original join column value and `null` replacements, as join column from left dataset
      *      appended column name builds using original left join column and `skewedColumnPostFix` separated by underscore.
      *
      * @note there is no checks how many `null` values on left dataset before applying above steps,
      *       as well as there is no checks does it sort merge join or broadcast.
      *
      * IMPORTANT: If left dataset already has appended column name, it will be reused to benefit already repartitioned data on the left
      *
      * HIGHLY IMPORTANT: right dataset should not contain negative values in `joinRightCol`
      */
    private[transformation] def nullSkewLeftJoin(right: DataFrame,
                                                 joinLeftCol: Column,
                                                 joinRightCol: Column,
                                                 skewedColumnPostFix: String = "skewed_column",
                                                 nullNumBuckets: Int = 10000): DataFrame = {

      val skewedTempColumn = s"${joinLeftCol.toString()}_$skewedColumnPostFix"

      if (underlying.columns.exists(_ equalsIgnoreCase skewedTempColumn)) {
        underlying.join(right.where(joinRightCol.isNotNull), col(skewedTempColumn) === joinRightCol, "left")
      } else {
        underlying
          .withColumn(skewedTempColumn,
                      when(joinLeftCol.isNotNull, joinLeftCol).otherwise(negativeRandomWithin(nullNumBuckets)))
          .join(right.where(joinRightCol.isNotNull), col(skewedTempColumn) === joinRightCol, "left")
      }
    }
  }

简而言之：我将左数据集连接键null值替换为负范围，以使其均匀地重新分区。

注意：此解决方案仅适用于左连接和nulljoin键倾斜。我不想爆炸正确的数据集，并做任何关键的倾斜解决方案。此外，在该步骤之后，nulljoin键值将被分配到不同的分区，因此，mapPartitions等将不起作用。

作为总结，上述解决方案对我有所帮助，但我希望看到更多针对此类数据集连接问题的解决方案。

Answer 2

我之前也遇到过同样的问题，但在做了一些性能测试之后，我选择了另一种方法。这取决于您的数据，数据将告诉您解决此连接问题的更好算法。

在我的例子中，我有超过30%的数据在join的左侧带有null，并且数据是拼花格式的。考虑到这一点，我最好执行一个过滤器，其中该键为null，并且该键不为null，只在不为null时连接，然后合并这两个数据。

val data = ...
val notJoinable = data.filter('keyToJoin.isNull)
val joinable = data.filter('keyToJoin.isNotNull)

joinable.join(...) union notJoinable

它也避免热点。如果我使用你的方法（负数/任何不是可连接的值），火花会洗牌所有这些数据，这是很多数据（超过30%）。

只是想给你展示解决问题的另一种方法，