pyspark - Pyspark 列的十分位数或其他分位数排名

Question

我有一个带有多个数字列的 pyspark DF，我想为每一列根据每个变量计算该行的十分位数或其他分位数等级。

这对 Pandas 来说很简单，因为我们可以使用 qcut 函数为每个变量创建一个新列，如 pd.qcut(x,q=n) 那样将 'q' 的值 0 分配给 n-1。

这如何在 pyspark 中完成？我尝试了以下操作，但显然断点在这些三分之一之间并不是唯一的。我想将数据的较低 1/3 分配为 1，将下一个 1/3 分配为 2，将前 1/3 分配为 3。我希望能够改变这一点，也许可以使用 1/10、1/32 等

w =  Window.partitionBy(data.var1).orderBy(data.var1)
d2=df.select(
    "var1",
     ntile(3).over(w).alias("ntile3")
)


agged=d2.groupby('ntile3').agg(F.min("var1").alias("min_var1"),F.max("var1").alias("max_var1"),F.count('*'))
agged.show()

+------+--------+--------+--------+
|ntile3|min_var1|max_var1|count(1)|
+------+--------+--------+--------+
|     1|     0.0|   210.0|  517037|
|     3|     0.0|   206.0|  516917|
|     2|     0.0|   210.0|  516962|
+------+--------+--------+--------+

Answer 1

QuantileDiscretizer来自 'pyspark.ml.feature'可以使用。

values = [(0.1,), (0.4,), (1.2,), (1.5,)]
df = spark.createDataFrame(values, ["values"])
qds = QuantileDiscretizer(numBuckets=2,
...     inputCol="values", outputCol="buckets", relativeError=0.01, handleInvalid="error")
bucketizer = qds.fit(df)
bucketizer.setHandleInvalid("skip").fit(df).transform(df).show()

+------+-------+
|values|buckets|
+------+-------+
|   0.1|    0.0|
|   0.4|    1.0|
|   1.2|    1.0|
|   1.5|    1.0|
+------+-------+