将具有最接近值的列添加到 PySpark Dataframe
Posted
技术标签:
【中文标题】将具有最接近值的列添加到 PySpark Dataframe【英文标题】:Add column with closest vaues to PySpark Dataframe 【发布时间】:2019-10-11 05:35:32 【问题描述】:我有一个 PySpark 数据框(比如 df
),它有两列(Name
和 Score
)。以下是数据框的示例:
+------+-----+
| Name|Score|
+------+-----+
| name1|11.23|
| name2|14.57|
| name3| 2.21|
| name4| 8.76|
| name5|18.71|
+------+-----+
我有一个 numpy 数组(比如 bin_array
),它的值接近 PySpark 数据框的标题为 Score
的列中的数值。
下面是前面提到的numpy数组:
bin_array = np.array([0, 5, 10, 15, 20])
我想将Score
列中每一行的值与bin_array
中的值进行比较,并将最接近的值(从bin_array
获取)存储在 PySpark 数据框中的单独列中。
下面是我希望我的新数据框(比如df_new
)的外观。
+------+-----+------------+
| Name|Score| Closest_bin|
+------+-----+------------+
| name1|11.23| 10.0 |
| name2|14.57| 15.0 |
| name3| 2.21| 0.0 |
| name4| 8.76| 10.0 |
| name5|18.71| 20.0 |
+------+-----+------------+
我有下面提到的函数,它给了我最接近 bin_array
的值。当我用单独的数字测试它时,该功能可以正常工作。
def find_nearest(array, value):
array = np.asarray(array)
idx = (np.abs(array - value)).argmin()
return float(array[idx])
在我的实际工作中,我会在datafrmae 中有数百万行。 创建df_new
的最有效方法是什么?
以下是我尝试用来创建用户定义函数 (udf) 和新数据框 (df_new
) 的步骤。
closest_bin_udf = F.udf( lambda x: find_nearest(array, x) )
df_new = df.withColumn( 'Closest_bin' , closest_bin_udf(df.Score) )
但是,当我尝试df_new.show()
时出现错误。部分错误如下所示。
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-11-685c9b7e25d9> in <module>()
----> 1 df_new.show()
/usr/lib/spark/python/pyspark/sql/dataframe.py in show(self, n, truncate, vertical)
376 """
377 if isinstance(truncate, bool) and truncate:
--> 378 print(self._jdf.showString(n, 20, vertical))
379 else:
380 print(self._jdf.showString(n, int(truncate), vertical))
您可以使用下面提到的步骤来创建上述数据框:
from pyspark.sql import *
import pyspark.sql.functions as F
import numpy as np
Stats = Row("Name", "Score")
stat1 = Stats('name1', 11.23)
stat2 = Stats('name2', 14.57)
stat3 = Stats('name3', 2.21)
stat4 = Stats('name4', 8.76)
stat5 = Stats('name5', 18.71)
stat_lst = [stat1 , stat2, stat3, stat4, stat5]
df = spark.createDataFrame(stat_lst)
df.show()
【问题讨论】:
【参考方案1】:您可以使用来自pyspark.mllib
的bucketizer
from pyspark.sql import *
import pyspark.sql.functions as F
import numpy as np
Stats = Row("Name", "Score")
stat_lst = [Stats('name1', 11.23) , Stats('name2', 14.57), Stats('name3', 2.21), Stats('name4', 8.76), Stats('name5', 18.71)]
df = spark.createDataFrame(stat_lst)
from pyspark.ml.feature import Bucketizer
"""
Bucketizer creates bins like 0-5:0, 5-10:1, 10-15:2, 15-20:3
As I see, your expected output wants the closest numbered bin, so you might
have to change your buckets or the variable `t` below accordingly.
"""
bucket_list = [0, 5, 10, 15, 20]
bucketizer = Bucketizer(splits=bucket_list, inputCol="Score", outputCol="buckets")
df_buck = bucketizer.setHandleInvalid("keep").transform(df)
df_buck.show()
我仍在努力寻找最近的垃圾箱,我会更新我的答案。
如果您想要每个存储桶的数组值,您可以使用 udf 创建一个包含存储桶名称的新列
from pyspark.sql.functions import udf
from pyspark.sql.types import *
t = dict(zip(range(len(bucket_list)), bucket_list))
udf_foo = udf(lambda x: t[x], IntegerType())
df_buck = df_buck.withColumn("score_bucket", udf_foo("buckets"))
输出
>>> df_buck.show()
+-----+-----+-------+------------+
| Name|Score|buckets|score_bucket|
+-----+-----+-------+------------+
|name1|11.23| 2.0| 10|
|name2|14.57| 2.0| 10|
|name3| 2.21| 0.0| 0|
|name4| 8.76| 1.0| 5|
|name5|18.71| 3.0| 15|
+-----+-----+-------+------------+
编辑:更正分数桶:
# Not dynamic, but please try to figure out this business logic according to your use-case
df_buck = df_buck.withColumn("correct_buckets", F.when(df_buck.Score-df_buck.score_bucket > 5/2, F.col("score_bucket") + 5).otherwise(F.col("score_bucket"))).drop("buckets", "score_bucket")
现在输出符合预期:
+-----+-----+---------------+
| Name|Score|correct_buckets|
+-----+-----+---------------+
|name1|11.23| 10|
|name2|14.57| 15|
|name3| 2.21| 0|
|name4| 8.76| 10|
|name5|18.71| 20|
+-----+-----+---------------+
【讨论】:
非常好的帖子@pisall:我能够很好地运行代码直到最后一行df_buck.show()
。当 df_buck.show()
尝试执行 df_buck = df_buck.withColumn("score_bucket", udf_foo("buckets"))
时,我收到以下错误:``` 19 df_buck = df_buck.withColumn( "kk_bucket", udf_nearest_bin("buckets") ) ---> 20 df_buck.show() /usr /lib/spark/python/pyspark/sql/dataframe.py in show(self, n, truncate, vertical) 377 if isinstance(truncate, bool) and truncate: --> 378 print(self._jdf.showString(n, 20, 垂直))```
@SiddharthSatpathy 无法理解,这是非常小的信息。我需要Py4JError
或AnalysisException
之后的错误。如果您喜欢使用编辑,可以将其添加到您的问题中
我能够解决这个错误。感谢这个很棒的答案。 =)【参考方案2】:
您也可以pandas_udf
,尽管我建议您在扩大规模时测试速度和内存消耗
from pyspark.sql.functions import pandas_udf, PandasUDFType
import numpy as np
import pandas as pd
df = spark.createDataFrame(zip(["name_"+str(i) for i in range(1,6)], [11.23, 14.57, 2.21, 8.76, 18.71]), ["Name", "Score"])
bin_array = np.array([0, 5, 10, 15, 20])
@pandas_udf('double', PandasUDFType.SCALAR)
def find_nearest(value):
res = bin_array[np.newaxis, :] - value.values[:, np.newaxis]
ret_vals = [bin_array[np.argmin(np.abs(i))] for i in res]
return pd.Series(ret_vals)
df.withColumn('v2', find_nearest(df.Score)).show()
输出
+------+-----+----+
| Name|Score| v2|
+------+-----+----+
|name_1|11.23|10.0|
|name_2|14.57|15.0|
|name_3| 2.21| 0.0|
|name_4| 8.76|10.0|
|name_5|18.71|20.0|
+------+-----+----+
【讨论】:
你有没有机会看到函数如何随着更大的数据进行缩放?以上是关于将具有最接近值的列添加到 PySpark Dataframe的主要内容,如果未能解决你的问题,请参考以下文章
如何使用pyspark将具有多个可能值的Json数组列表转换为数据框中的列