如何使用 partitionedBy 计算按列分组的百分位数？答案

【问题标题】：How to calculate percentiles grouped by column using partitionedBy?如何使用 partitionedBy 计算按列分组的百分位数？
【发布时间】：2020-03-13 11:09:37
【问题描述】：

我正在使用 spark-sql-2.4.1v，我正在尝试查找分位数，即在我给定数据的每一列上，百分位 0、百分位 25 等。当我在做多个百分位数时，如何检索每个计算的结果的百分位数？

我的数据框df：

+----+---------+-------------+----------+-----------+-------+
|  id|     date|      revenue|con_dist_1| con_dist_2| zone  |
+----+---------+-------------+----------+-----------+-------+
|  10|1/15/2018|  0.010680705|        10|0.019875458|  east |
|  10|1/15/2018|  0.006628853|         4|0.816039063|  west |
|  10|1/15/2018|   0.01378215|        20|0.082049528|  east |
|  10|1/15/2018|  0.010680705|         6|0.019875458|  west |
|  10|1/15/2018|  0.006628853|        30|0.816039063|  east |
+----+---------+-------------+----------+-----------+-------+

最终的数据框应如下所示，即对于每个区域：

+---+---------+-----------+-------+-------------+-----------+-----------+
| id|     date|    revenue|  zone | perctile_col| quantile_0|quantile_10|
+---+---------+-----------+-------+-------------+-----------+-----------+
| 10|1/15/2018|0.010680705|  east |  con_dist_1 |       10.0|       30.0|
| 10|1/15/2018|0.010680705|  east |  con_dist_2 |0.019875458|0.816039063|
| 10|1/15/2018|0.010680705|  west |  con_dist_1 |        4.0|        6.0|
| 10|1/15/2018|0.010680705|  west |  con_dist_2 |0.019875458|0.816039063|
+---+---------+-----------+-------+-------------+-----------+-----------+

有什么方法可以使用partitionBy 和approxQuantile 函数吗？是否会使用repartition("zone") 进行处理，即不收集每个区域的数据集？

【问题讨论】：

您打算如何处理此处的收入列（因为结果只有 2 列）？根据文本，预期结果是否与此处的结果类似：stackoverflow.com/questions/60561513/… 但行数是原来的两倍（因为有 2 个区域）？

标签： java scala apache-spark apache-spark-sql quantile

【解决方案1】：

approxQuantile 在这里不太适合，因为它不允许分组。然而，这个问题可以使用 percentile_approx 和 Spark 窗口函数来解决（groupBy 也可以在这里使用，使用哪一个取决于所需的数据帧格式）。首先我们做一些设置：

val df = Seq(
    (10, "1/15/2018", 0.010680705, 10,0.019875458, "east"),
    (10, "1/15/2018", 0.006628853,  4,0.816039063, "west"),
    (10, "1/15/2018", 0.01378215,  20,0.082049528, "east"),
    (10, "1/15/2018", 0.010680705,  6,0.019875458, "west"),
    (10, "1/15/2018", 0.006628853, 30,0.816039063, "east"))     
  .toDF("id", "date", "revenue", "con_dist_1", "con_dist_2", "zone")


val percentiles = Seq(0.1, 1.0)  // Which percentiles to calculate
val cols = Seq("con_dist_1", "con_dist_2")  // The columns to use

要计算每个区域组的百分位数，可以按如下方式完成：

val window = Window.partitionBy("zone")
val percentile_func = (col: String) => expr(s"percentile_approx(${col}, array(${percentiles.mkString(",")}))")
val df2 = cols.foldLeft(df){case (df, c) => df.withColumn(c, percentile_func(c).over(window))}

结果会是这样的：

+---+---------+-----------+----------+--------------------------+----+
|id |date     |revenue    |con_dist_1|con_dist_2                |zone|
+---+---------+-----------+----------+--------------------------+----+
|10 |1/15/2018|0.006628853|[4, 6]    |[0.019875458, 0.816039063]|west|
|10 |1/15/2018|0.010680705|[4, 6]    |[0.019875458, 0.816039063]|west|
|10 |1/15/2018|0.010680705|[10, 30]  |[0.019875458, 0.816039063]|east|
|10 |1/15/2018|0.01378215 |[10, 30]  |[0.019875458, 0.816039063]|east|
|10 |1/15/2018|0.006628853|[10, 30]  |[0.019875458, 0.816039063]|east|
+---+---------+-----------+----------+--------------------------+----+

接下来，我们要将数据帧转换为正确的格式。这是对这里答案的轻微改编：How to include/map calculated percentiles to the result dataframe?。

cols.map{ case c =>
  percentiles
    .zipWithIndex
    .foldLeft(df2.withColumn("perctile_col", lit(c))){ case (df2, (perc, index)) => 
      df2.withColumn(s"qunantile_${perc}", col(c).getItem(index))
    }
  }
  .reduce(_.union(_))
  .drop(cols: _*) // these are not needed anymore

最终数据框：

+---+---------+-----------+----+------------+-------------+-------------+
| id|     date|    revenue|zone|perctile_col|qunantile_0.1|qunantile_1.0|
+---+---------+-----------+----+------------+-------------+-------------+
| 10|1/15/2018|0.006628853|west|  con_dist_1|          4.0|          6.0|
| 10|1/15/2018|0.010680705|west|  con_dist_1|          4.0|          6.0|
| 10|1/15/2018|0.010680705|east|  con_dist_1|         10.0|         30.0|
| 10|1/15/2018| 0.01378215|east|  con_dist_1|         10.0|         30.0|
| 10|1/15/2018|0.006628853|east|  con_dist_1|         10.0|         30.0|
| 10|1/15/2018|0.006628853|west|  con_dist_2|  0.019875458|  0.816039063|
| 10|1/15/2018|0.010680705|west|  con_dist_2|  0.019875458|  0.816039063|
| 10|1/15/2018|0.010680705|east|  con_dist_2|  0.019875458|  0.816039063|
| 10|1/15/2018| 0.01378215|east|  con_dist_2|  0.019875458|  0.816039063|
| 10|1/15/2018|0.006628853|east|  con_dist_2|  0.019875458|  0.816039063|
+---+---------+-----------+----+------------+-------------+-------------+

【讨论】：

@BdEngineer：我应该可以再仔细看看。但是，快速浏览一下，如果您在问题中添加了两个数据帧的预期输出，将会有所帮助。
@BdEngineer：第二点，这个答案是否有助于您解决这里的问题？ :)
@BdEngineer：foldLeft 部分实际上只是多个withColumn，但输入的列不同，所以它不应该太慢（并且直接在执行程序上完成）。
@BdEngineer: withColumn 只会将信息添加到数据框（不像选择）。 foldLeft 的输入是 df2.withColumn("perctile_col", lit(c))，这是一个包含所有相关列的数据框。
@BdEngineer：您能否将数据和所有添加信息（是否应该针对每个组或整个数据集/多个百分位数/等）添加到新问题中？在 cmets 中回答有点过分了。