Spark - 如何计算Spark中的百分位数？

Question

我试图获得单列数据帧的0.8百分位数.我试过这样的方式:

val limit80 = 0.8
val dfSize = df.count()
val perfentileIndex = dfSize*limit80 

dfSorted = df.sort()
val percentile80 = dfSorted .take(perfentileIndex).last()

但我认为这对大型数据帧来说是失败的,因为它们可能分布在不同的节点上.

有没有更好的方法来计算百分位数？或者我怎么能在同一台机器中拥有数据帧的所有行(即使这是非常反模式的),所以这df.take(index)将真正考虑整个数据集而不仅仅是节点中的分区.

Answer 1

对于Spark 2.x,您可以使用approxQuantile,如下例所示:

val df = Seq(
  10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
  20, 21, 22, 23, 24, 25, 26, 27, 28, 29
).toDF("num")

df.stat.approxQuantile("num", Array(0.8), 0.1)
// res4: Array[Double] = Array(26.0)

请注意,第3个参数越小relativeError,计算越昂贵.以下是API文档中的相关说明:

relativeError:要达到的相对目标精度(大于或等于0).如果设置为零,则计算精确的分位数,这可能非常昂贵.

Answer 2

您可以使用 Spark SQL 函数approx_percentile(col, percentage)：

val df = Seq(0.5, 0.4, 0.1).toDF
df.agg(expr("approx_percentile(value, array(0.5))").as("percentile")).show
// +----------+
// |percentile|
// +----------+
// |     [0.4]|
// +----------+

https://spark.apache.org/docs/latest/api/sql/index.html#approx_percentile