Ableitung von Perzentil der Verwendung von Spark-Daten-frame und GroupBy in python

Weiß ich eine Lösung, um die Perzentil jede Zeile mit RDDs. Zuerst konvertieren Sie Ihre RDD, um einen DataFrame:

# convert to rdd of dicts
rdd = df.rdd
rdd = rdd.map(lambda x: x.asDict())

Dann können Sie berechnen jede Zeile Perzentil:

column_to_decile = 'price'
total_num_rows = rdd.count()


def add_to_dict(_dict, key, value):
    _dict[key] = value
    return _dict


def get_percentile(x, total_num_rows):
    _dict, row_number = x
    percentile = x[1] / float(total_num_rows)
    return add_to_dict(_dict, "percentile", percentile)


rdd_percentile = rdd.map(lambda d: (d[column_to_decile], d)) # make column_to_decile a key
rdd_percentile = rdd_percentile.sortByKey(ascending=False) # so 1st decile has largest
rdd_percentile = rdd_percentile.map(lambda x: x[1]) # remove key
rdd_percentile = rdd_percentile.zipWithIndex() # append row number
rdd_percentile = rdd_percentile.map(lambda x: get_percentile(x, total_num_rows))

Und schließlich wieder zurück konvertieren in ein DataFrame mit:

df = sqlContext.createDataFrame(rdd_percentile)

Um die Zeile mit dem nächsten Perzentil 0,6, könnte man etwas wie das hier tun:

from pyspark.sql.types import *
from pyspark.sql.functions import udf


def get_row_with_percentile(df, percentile):
    func = udf(lambda x: abs(x), DoubleType())
    df_distance = df.withColumn("distance", func(df['percentile'] - percentile))
    min_distance = df_distance.groupBy().min('distance').collect()[0]['min(distance)']
    result = df_distance.filter(df_distance['distance'] == min_distance)
    result.drop("distance")
    return result


get_row_with_percentile(df, 0.6).show()

InformationsquelleAutor Galen Long