pyspark中的累积百分位数计算[重复]答案

【问题标题】：cumulative percentile calculation in pyspark [duplicate]pyspark中的累积百分位数计算[重复]
【发布时间】：2021-10-18 05:30:38
【问题描述】：

我有一个这样的数据框：

|SEQ_ID |TIME_STAMP             |RESULT          |
+-------+-----------------------+----------------+
|3879859|2021-08-31 19:54:53.88 |25.9485244750994|
|3879859|2021-08-31 21:16:06.228|35.9163284302007|
|3879859|2021-08-31 22:28:46.306|41.9778823852006|
|3879859|2021-08-31 22:28:46.306|41.9778823852006|
|3879859|2021-08-31 23:12:08.058|39.9112701415998|
|3879859|2021-08-31 23:17:35.796|33.0476760864009|
|3879859|2021-08-31 23:47:17.383|60.2846145630007|
|3879859|2021-09-01 00:00:26.722|67.0690536499006|
|3879859|2021-09-01 00:00:26.722|67.0690536499006|
|3879859|2021-09-01 00:02:07.825|67.8424835205007|

对于 pyspark 中的正常百分位数计算，我使用以下内容：

df.groupBy('SEQ_ID')\
.agg(f.expr('percentile(RESULT, 0.25)').alias('Q1'),
     f.expr('percentile(RESULT, 0.50)').alias('Median'),
     f.expr('percentile(RESULT, 0.75)').alias('Q3'))\

但这需要按SEQ_ID 分组的所有数据。我想使用上面的行计算每行的 Q1、中位数和 Q3。

|SEQ_ID |TIME_STAMP             |RESULT          |Q1   |Median|Q3   |
+-------+-----------------------+----------------+-----+------+-----+
|3879859|2021-08-31 19:54:53.88 |25.9485244750994|
|3879859|2021-08-31 21:16:06.228|35.9163284302007|
|3879859|2021-08-31 22:28:46.306|41.9778823852006|
|3879859|2021-08-31 22:28:46.306|41.9778823852006|
|3879859|2021-08-31 23:12:08.058|39.9112701415998|
|3879859|2021-08-31 23:17:35.796|33.0476760864009|
|3879859|2021-08-31 23:47:17.383|60.2846145630007|
|3879859|2021-09-01 00:00:26.722|67.0690536499006|
|3879859|2021-09-01 00:00:26.722|67.0690536499006|
|3879859|2021-09-01 00:02:07.825|67.8424835205007|

所以第一行的Q1、Median 和Q3 将是：25.9485244750994 对于第二行，百分位数将使用25.9485244750994 和35.9163284302007 等等来计算。

如果我定义一个这样的窗口

w=Window.partitionBy('SEQ_ID').orderBy(col('TIME_STAMP').asc()).rangeBetween(Window.unboundedPreceding,0)

下面的代码可以工作吗？：

df.groupBy('SEQ_ID')\
.agg(f.expr('percentile(Pad_Wear, 0.25)').alias('Q1'),
     f.expr('percentile(Pad_Wear, 0.50)').alias('Median'),
     f.expr('percentile(Pad_Wear, 0.75)').alias('Q3')).over(w)

【问题讨论】：

@ggordon 谢谢。是的，它确实。这几乎是我的预期。但是，我低估了累积效应，我将percentile 更改为percentile_approx，因为我有一个非常大的数据集。

标签： apache-spark pyspark window iterated-function

【解决方案1】：

您可以使用monotonically_increasing_id 创建一个新的“假 ID 列”，然后按它分组或按它窗口化，它将逐行运行

【讨论】：

感谢您的建议。我考虑过使用它，但它会将整个数据集加载到驱动程序节点并创建大量随机写入。我尝试使用row_number()，但它仍然很慢。 window 函数似乎更快。虽然它仍然比我刚刚使用 groupBy 的时间长了 50 分钟