这是数据倾斜的常见问题，您可以采取多种方法。

如果倾斜随着时间的推移保持稳定，列表分桶就可以工作，这可能是也可能不是，特别是如果引入了分区变量的新值。我没有研究过随着时间的推移调整列表分桶有多容易，正如你的评论所说，你无论如何都不能使用它，因为它是Spark 2.0的特性。

如果您使用的是1.6. x，关键是您可以创建自己的函数，将每个事件名称映射到600个唯一值之一。您可以将其作为UDF或大小写表达式来执行。然后，您只需使用该函数创建一个列，然后使用重新分区（600，'myPARtionCol）按该列分区，而不是合并（600）。

因为我们在Swoop处理非常倾斜的数据，所以我发现以下主力数据结构对于构建与分区相关的工具非常有用。

/** Given a key, returns a random number in the range [x, y) where
  * x and y are the numbers in the tuple associated with a key.
  */
class RandomRangeMap[A](private val m: Map[A, (Int, Int)]) extends Serializable {
  private val r = new java.util.Random() // Scala Random is not serializable in 2.10

  def apply(key: A): Int = {
    val (start, end) = m(key)
    start + r.nextInt(end - start)
  }

  override def toString = s"RandomRangeMap($r, $m)"
}

例如，以下是我们如何为略有不同的情况构建分区器：数据倾斜且键数量较少，因此我们必须增加倾斜键的分区数，同时坚持使用1作为每个键的最小分区数：

/** Partitions data such that each unique key ends in P(key) partitions.
  * Must be instantiated with a sequence of unique keys and their Ps.
  * Partition sizes can be highly-skewed by the data, which is where the
  * multiples come in.
  *
  * @param keyMap  maps key values to their partition multiples
  */
class ByKeyPartitionerWithMultiples(val keyMap: Map[Any, Int]) extends Partitioner {
  private val rrm = new RandomRangeMap(
    keyMap.keys
      .zip(
        keyMap.values
          .scanLeft(0)(_+_)
          .zip(keyMap.values)
          .map {
            case (start, count) => (start, start + count)
          }
      )
      .toMap
  )

  override val numPartitions =
    keyMap.values.sum

  override def getPartition(key: Any): Int =
    rrm(key)
}

object ByKeyPartitionerWithMultiples {

  /** Builds a UDF with a ByKeyPartitionerWithMultiples in a closure.
    *
    * @param keyMap  maps key values to their partition multiples
    */
  def udf(keyMap: Map[String, Int]) = {
    val partitioner = new ByKeyPartitionerWithMultiples(keyMap.asInstanceOf[Map[Any, Int]])
    (key:String) => partitioner.getPartition(key)
  }

}

在您的情况下，您必须将多个事件名称合并到一个分区中，这需要更改，但我希望上面的代码能让您了解如何解决问题。

最后一个观察是，如果随着时间的推移，事件名称在数据中的分布值很大，您可以执行统计信息收集，传递部分数据以计算映射表。您不必一直这样做，只需在需要时进行。为了确定这一点，您可以查看每个分区中输出文件的行数和/或大小。换句话说，整个过程可以作为Spark作业的一部分自动化。