Percentile rank in pyspark using QuantileDiscretizer

Question

I am wondering if it's possible to obtain the result of percentile_rank using the QuantileDiscretizer transformer in pyspark.

The purpose is that I am trying to avoid computing the percent_rank over the entire column, as it generates the following error:

WARN WindowExec: No Partition Defined for Window operation! 
Moving all data to a single partition, this can cause serious performance degradation.

The method I am following is to first use QuantileDiscretizer then normalize to [0,1]:

from pyspark.sql.window import Window
from pyspark.ml.feature import QuantileDiscretizer
from scipy.stats import gamma

X1 = gamma.rvs(0.2, size=1000)

df = spark.createDataFrame(pd.DataFrame(X1, columns=["x"]))

df = df.withColumn("perc_rank", F.percent_rank().over(Window.orderBy("x")))
df = QuantileDiscretizer(numBuckets=df.count()+1,\
                         inputCol="x",\
                         outputCol="q_discretizer").fit(df).transform(df)

agg_values = df.agg(F.max(df["q_discretizer"]).alias("maxval"),\
                    F.min(df["q_discretizer"]).alias("minval")).collect()[0]

xmax, xmin = agg_values.__getitem__("maxval"), agg_values.__getitem__("minval")
normalize = F.udf(lambda x: (x-xmin)/(xmax-xmin))

df = df.withColumn("perc_discretizer", normalize("q_discretizer"))
df = df.withColumn("error", F.round(F.abs(F.col("perc_discretizer")- F.col("perc_rank")),6) )
print(df.select(F.max("error")).show())
df.show(5)

However, it seems that increasing the number of datapoints the error grows, so I am not sure this is the right way to do it.

Is it possible to use QuantileDiscretizer to obtain the percentile_rank?

Alternatively is there a way to compute percentile_rank over an entire column in an efficient way?

user_7562895 · Answer

Well you can use the below to avoid the warning message:

X1 = gamma.rvs(0.2, size=10)
df = spark.createDataFrame(pd.DataFrame(X1, columns=["x"]))
df = df.withColumn("dummyCol", F.lit("some_val"))
win = Window.partitionBy("dummyCol").orderBy("x")
df = df.withColumn("perc_rank", F.percent_rank().over(win)).drop("dummyCol")

but nonetheless, the data would still be moved to a single worker, I don't think so there is any better alternative to avoid the shuffle here since the complete column needs to be rank-ordered.

In case you have multiple windows over the same column, you can try to pre-partition the data and then apply the ranking functions.

Percentile rank in pyspark using QuantileDiscretizer

1 Answers