PySpark按预测值排序后筛选至标签累积均值最后一次低于阈值

现有一个包含predictions列与二值labels列的DataFrame，其中labels列中1的占比约99%、0的占比约1%，且predictions与labels存在正相关关系（按predictions排序后划分十分位，各分位的labels均值呈递增趋势）。

需要完成以下操作：

• 按predictions对DataFrame进行排序
• 筛选排序后的DataFrame，保留到labels累积均值最后一次低于阈值的所有行——即使中间出现均值突破阈值后又回落的情况，回落部分的行也需纳入筛选结果

示例（实际无running_mean列） #

preds   labels  running_mean
0   0.05    1   1.000000
1   0.10    0   0.500000
2   0.15    0   0.333333
3   0.20    0   0.250000
4   0.25    1   0.400000
5   0.30    0   0.333333
6   0.35    0   0.285714
7   0.40    1   0.375000
8   0.45    1   0.444444
9   0.50    1   0.500000
10  0.55    1   0.545455
11  0.60    1   0.583333
12  0.65    1   0.615385
13  0.70    1   0.642857
14  0.75    0   0.600000
15  0.80    1   0.625000
16  0.85    1   0.647059
17  0.90    1   0.666667
18  0.95    1   0.684211

当阈值设为0.63时，需保留前16行（包含索引为15的行）。

内容的提问来源于stack exchange，提问作者figs_and_nuts