pyspark 1.6 中 pandas 分位数和 cut 的替代方法有哪些

Question

我是 pyspark 的新手。我有如下的熊猫代码。

bindt = df[df[var].notnull()][var].quantile([0,.1,.2,.3,.4,.5,.6,.7,.8,.9,1]).unique()

df['{0}_quartile'.format(var)] = pd.cut(df[var], bindt, labels=False, include_lowest=True )

我在 pyspark 2.x 中找到了“approxQuantile”，但在 pyspark 1.6.0 中没有找到任何此类内容

我的示例输入：

df.show()

+-----------+----------+---------------+--------------+------------------------+
|  id       | col_1    |col_2          |col_3         |col_4                   |
+-----------+----------+---------------+--------------+------------------------+
|1.10919E+16|3988487.35|     -236751.43|    -362208.07|                0.660000|
|1.10919E+16|3988487.35|     -236751.43|    -362208.07|                0.900000|
|1.10919E+16|3988487.35|     -236751.43|    -362208.07|                0.660000|
|1.10919E+16|  36718.55|           null|          null|                0.860000|
|1.10919E+16|  36718.55|           null|          null|                0.780000|
|1.10919E+16|  36718.55|           null|          null|                0.660000|
|1.10919E+16|  36718.55|           null|          null|                0.900000|
|1.10919E+16|  36718.55|           null|          null|                0.660000|

df.collect()

[Row(id=u'1.11312E+16', col_1=Decimal('367364.44'), col_2=Decimal('-401715.23'), col_3=Decimal('-1649917.53'), col_4=Decimal('0.080000')), 
Row(id=u'1.11312E+16', col_1=Decimal('367364.44'), col_2=Decimal('-401715.23'), col_3=Decimal('-1649917.53'), col_4=Decimal('0.780000')), 
Row(id=u'1.11312E+16', col_1=Decimal('367364.44'), col_2=Decimal('-401715.23'), col_3=Decimal('-1649917.53'), col_4=Decimal('0.780000')), 
Row(id=u'1.11312E+16', col_1=Decimal('367364.44'), col_2=Decimal('-401715.23'), col_3=Decimal('-1649917.53'), col_4=Decimal('0.860000')), 
Row(id=u'1.11312E+16', col_1=Decimal('367364.44'), col_2=Decimal('-401715.23'), col_3=Decimal('-1649917.53'), col_4=Decimal('0.330000'))]

我必须对所有输入列循环上述逻辑。

for var in df.columns:
    bindt = df[df[var].notnull()][var].quantile([0,.1,.2,.3,.4,.5,.6,.7,.8,.9,1]).unique()    
    df['{0}_quartile'.format(var)] = pd.cut(df[var], bindt, labels=False, include_lowest=True )

任何人都可以建议如何在 pyspark 1.6 dataframe 中重写上述代码。

提前致谢

Answer 1

如果您使用的是 pyspark 2.x，则可以使用ml lib 中的QuantileDiscretizer ，它在底层使用了approxQuantile()和Bucketizer 。

\n\n

但是，由于您使用的是 pyspark 1.6.x，您需要：

\n\n

1. 查找列的分位数值

\n\n

您可以通过两种方式找到分位数值：

\n\n

\n

1. 通过计算percent_rank()来计算列的百分位数，并提取百分位数值接近所需分位数的列值

\n

2. 请遵循此答案中的方法，该答案解释了如何使用 pyspark < 2.0.0 执行分位数近似

\n

\n\n

这是我的近似分位数值的示例实现：

\n\n

from pyspark.sql import functions as F\nfrom pyspark.sql import Window\n\ndef compute_quantiles(df, col, quantiles):\n  quantiles = sorted(quantiles)\n\n  # 1. compute percentile\n  df = df.withColumn("percentile", F.percent_rank().over(Window.orderBy(col)))\n\n  # 2. categorize quantile based on the desired quantile and compute errors\n  df = df.withColumn("percentile_cat1", F.lit(-1.0))\n  df = df.withColumn("percentile_err1", F.lit(-1.0))\n  df = df.withColumn("percentile_cat2", F.lit(-1.0))\n  df = df.withColumn("percentile_err2", F.lit(-1.0))\n\n  # check percentile with the lower boundaries\n  for idx in range(0, len(quantiles)-1):\n    q = quantiles[idx]\n    df = df.withColumn("percentile_cat1", F\\\n                       .when( (F.col("percentile_cat1") == -1.0) & \n                             (F.col("percentile") <= q), q)\\\n                       .otherwise(F.col("percentile_cat1")))\n    df = df.withColumn("percentile_err1", F\\\n                       .when( (F.col("percentile_err1") == -1.0) & \n                             (F.col("percentile") <= q), \n                             F.pow(F.col("percentile") - q, 2))\\\n                       .otherwise(F.col("percentile_err1")))\n\n  # assign the remaining -1 values in the error to the largest squared error of 1\n  df = df.withColumn("percentile_err1", F\\\n                     .when(F.col("percentile_err1") == -1.0, 1)\\\n                     .otherwise(F.col("percentile_err1")))\n\n  # check percentile with the upper boundaries\n  for idx in range(1, len(quantiles)):\n    q = quantiles[idx]\n    df = df.withColumn("percentile_cat2", F\\\n                       .when((F.col("percentile_cat2") == -1.0) & \n                             (F.col("percentile") <= q), q)\\\n                       .otherwise(F.col("percentile_cat2")))\n    df = df.withColumn("percentile_err2",F\\\n                       .when((F.col("percentile_err2") == -1.0) & \n                             (F.col("percentile") <= q), \n                             F.pow(F.col("percentile") - q, 2))\\\n                       .otherwise(F.col("percentile_err2")))\n\n  # assign the remaining -1 values in the error to the largest squared error of 1\n  df = df.withColumn("percentile_err2", F\\\n                     .when(F.col("percentile_err2") == -1.0, 1)\\\n                     .otherwise(F.col("percentile_err2")))\n\n  # select the nearest quantile to the percentile\n  df = df.withColumn("percentile_cat", F\\\n                     .when(F.col("percentile_err1") < F.col("percentile_err2"), \n                           F.col("percentile_cat1"))\\\n                     .otherwise(F.col("percentile_cat2")))\n  df = df.withColumn("percentile_err", F\\\n                     .when(F.col("percentile_err1") < F.col("percentile_err2"), \n                           F.col("percentile_err1"))\\\n                     .otherwise(F.col("percentile_err2")))\n\n  # 3. approximate quantile values by choosing the value with the lowest error at each percentile category\n  df = df.withColumn("approx_quantile", F\\\n                     .first(col).over(Window\\\n                                      .partitionBy("percentile_cat")\\\n                                      .orderBy(F.asc("percentile_err"))))\n\n  return df\n\ndef extract_quantiles(df):\n  df_quantiles = df.select("percentile_cat", "approx_quantile").distinct()\n  rows = df_quantiles.collect()\n  quantile_values = [ row.approx_quantile for row in rows ]\n\n  return quantile_values\n

\n\n

我想要从上面实现的是计算列中每行的百分位数，并将其分类到最近的分位数。将百分位数分类到最接近的分位数可以通过选择与百分位数差异（平方误差）最小的最低分位数类别来完成。

\n\n 1. 计算百分位\n\n

首先，我使用pyspark 中的Window 函数Percent_rank ()计算列的百分位。您可以将 Window 视为数据的分区规范。由于是Window函数，所以需要传入Window。percent_rank()

\n\n 2. 将百分位数分类到分位数边界并计算误差\n\n

最接近百分位数的分位数类别可以低于、等于或高于该百分位数。因此，我需要计算误差两次：第一次将百分位数与下分位数界限进行比较，第二次将其与上分位数界限进行比较。请注意，\xe2\x89\xa4 运算符用于检查百分位数是否小于或等于边界。在知道百分位数的直接上分位数边界和下分位数边界后，我们可以通过选择误差最低的低于或等于或高于或等于类别的分位数，将百分位数分配给最近的分位数类别。

\n\n 3. 近似分位数值\n\n

一旦我们知道每个百分位数的所有最接近的分位数类别，我们就可以近似分位数值：它是每个分位数类别具有最低误差的值。这个近似分位数值可以使用Windowfirst()在每个类别分区处使用函数来计算。接下来，要提取分位数值，我们只需从数据框中选择唯一的percentileCategory-approxQuantileValue 对。

\n\n\n\n

使用 测试我的数据（~10000 行）后desired_quantiles = [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]，我发现我的示例实现与approxQuantile结果非常接近。当我减少提供给 的误差时，两个结果值变得更加接近approxQuantile。

\n\n

使用extract_quantiles(compute_quantile(df, col, quantiles))：

\n\n

\n\n

使用approxQuantile：

\n\n

\n\n

2.使用分桶器

\n\n

找到分位数值后，您可以使用 pyspark 的 Bucketizer 根据分位数对值进行分桶。Bucketizer 在 pyspark 1.6.x [1] [2]和 2.x [3] [4]中均可用

\n\n

以下是如何执行分桶化的示例：

\n\n

from pyspark.ml.feature import Bucketizer\n\nbucketedData = df\ndesired_quantiles = [0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0] # must be sorted\n\nfor col in df.columns:\n  quantile_values = extract_quantiles(compute_quantiles(df, col, desired_quantiles))\n  splits = [ boundary_values ] # replace this with quantile_values\n\n  bucketizer = Bucketizer()\\\n    .setInputCol(col)\\\n    .setOutputCol("{}_quantile".format(col))\\\n    .setSplits(splits)\n  bucketedData = bucketizer.transform(bucketedData)\n

\n\n

您可以替换value_boundaries为您在步骤 1 中找到的分位数值或您想要的任何存储桶分割范围。当您使用 Bucketizer 时，整个列值范围必须包含在 split 内。否则，指定分割之外的值将被视为错误。如果您不确定数据的值边界-float("inf")，则必须显式提供无限值（例如 , ）以覆盖所有浮动值。float("inf")

\n