什么是按列分区但保持固定分区计数的有效方法？

Question

按字段将数据划分为预定义分区计数的最佳方法是什么？

我目前通过指定partionCount = 600来分区数据.发现计数600为我的数据集/集群设置提供了最佳查询性能.

val rawJson = sqlContext.read.json(filename).coalesce(600)
rawJson.write.parquet(filenameParquet)

现在我想通过列'eventName'对这些数据进行分区,但仍然保持计数600.数据当前有大约2000个唯一的eventNames,加上每个eventName中的行数不统一.大约10个eventNames有超过50%的数据导致数据偏斜.因此,如果我像下面那样进行分区,它的性能不是很好.写入比没有写入时间长5倍.

val rawJson = sqlContext.read.json(filename)
rawJson.write.partitionBy("eventName").parquet(filenameParquet)

为这些场景划分数据的好方法是什么？有没有办法按eventName进行分区,但是将其扩展到600个分区？

我的架构如下所示:

{  
  "eventName": "name1",
  "time": "2016-06-20T11:57:19.4941368-04:00",
  "data": {
    "type": "EventData",
    "dataDetails": {
      "name": "detailed1",
      "id": "1234",
...
...
    }
  }
} 

谢谢!

Answer 1

这是数据偏斜的常见问题,您可以采取多种方法.

如果偏斜随时间保持稳定,则列表分段工作,这可能是也可能不是,特别是如果引入了分区变量的新值.我没有研究随着时间的推移调整列表存储是多么容易,并且正如您的注释所述,无论如何都不能使用它,因为它是Spark 2.0功能.

如果你使用1.6.x,关键的观察是你可以创建自己的函数,将每个事件名称映射到600个唯一值之一.您可以将其作为UDF或案例表达式执行.然后,您只需使用该函数创建一个列,然后使用该列进行分区,repartition(600, 'myPartitionCol)而不是coalesce(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作业的一部分自动化.