UDF的性能改进-在pyspark中获得每行最小值的列名

如何解决UDF的性能改进-在pyspark中获得每行最小值的列名

我使用这个udf：

mincol = F.udf(lambda row: cols[row.index(min(row))],StringType())
df = df.withColumn("mycol",mincol(F.struct([df[x] for x in cols])))

获取每行最少的列名称作为另一个名为“ mycol”的列的值。

但是此代码非常慢。 有任何改善性能的建议吗？ 我正在使用Pyspark 2.3

解决方法

这里是Spark 2.3的另一种解决方案，它仅使用内置函数：

Caused by: java.lang.ArrayIndexOutOfBoundsException: 16227

步骤：

1. 使用可能的浮点数填充所有from sys import float_info from pyspark.sql.functions import array,least,col,lit,concat_ws,expr cols = df.columns col_names = array(list(map(lit,cols))) set_cols = list(map(col,cols)) # replace null with largest python float df.na.fill(float_info.max) \ .withColumn("min",least(*cols)) \ .withColumn("cnames",col_names) \ .withColumn("set",concat_ws(",",*set_cols)) \ .withColumn("min_col",expr("cnames[find_in_set(min,set) - 1]")) \ .select(*[cols + ["min_col"]]) \ .show() 。由于很难找到更大的值，因此这是空值替换的理想选择。
2. 使用least查找分钟列。
3. 创建列null来存储列名称。
4. 创建列cnames，其中包含所有值，以逗号分隔的字符串。
5. 使用find_in_set创建列set。该函数分别处理每个字符串项，并将返回找到的项的索引。最后，我们使用带有min_col的索引来检索列名。

,

这里是没有udf的方法。这个想法是创建一个包含每个列的值和名称的数组，然后对该数组进行排序。

df1 = spark.createDataFrame([
    (1.,2.,3.),(3.,1.),(9.,8.,-1.),(1.2,1.2,9.1),None,1.0)],\
        ["col1","col2","col3"])

cols = df1.columns
col_string = ','.join("'{0}'".format(c) for c in cols)

df1 = df1.withColumn("vals",F.array(cols)) \
    .withColumn("cols",F.expr("Array(" + col_string + ")")) \
    .withColumn("zipped",F.arrays_zip("vals","cols")) \
    .withColumn("without_nulls",F.expr("filter(zipped,x -> not x.vals is null)")) \
    .withColumn("sorted",F.expr("array_sort(without_nulls)")) \
    .withColumn("min",F.col("sorted")[0].cols) \
    .drop("vals","cols","zipped","without_nulls","sorted")
df1.show(truncate=False)

打印

+----+----+----+----+                                                           
|col1|col2|col3|min |
+----+----+----+----+
|1.0 |2.0 |3.0 |col1|
|3.0 |2.0 |1.0 |col3|
|9.0 |8.0 |-1.0|col3|
|1.2 |1.2 |9.1 |col1|
|3.0 |null|1.0 |col3|
+----+----+----+----+