Databricks 火花 UDF 不适用于过滤的数据帧

Posted 2023-04-13

【中文标题】Databricks 火花 UDF 不适用于过滤的数据帧

【英文标题】：Databricks spark UDF not working on filtered dataframe

【发布时间】：2020-11-26 15:43:49

【问题描述】：

我在使用 Pyspark 的 Databricks 中遇到了一个问题，如果我在这里遗漏了一些概念性的东西，我正试图了解为什么这个实现不起作用。 我要做的是在数据框中的列上运行 UDF，但只能在非空值上运行。

如果我将 lstrip_udf 调用替换为“Val123”之类的固定值，那么它可以正常工作，但它不适用于 UDF。如果我在 UDF 中使用不同的实现方式实现空值检查，那么它也可以工作。 但即使有 when 和 IsNotNull 它仍然会抛出以下错误。

有人可以解释为什么或我在这里缺少什么来完成这项工作吗？

代码：

from pyspark.sql.types import StructType, StructField, IntegerType, StringType, BooleanType, TimestampType
inputschema = StructType([StructField("testcol", StringType(), True),
                          StructField("testcol2", StringType(), True)
                         ]
                        )
inputfile = spark.createDataFrame([("012121212","Ref #1"),
                                   ("0034343434","Ref #2"),
                                   ("0034343434","Ref #3"),
                                   (None,"Ref #4"),
                                   (None,"Ref #5"),
                                   ("00998877","Ref #6")
                                  ],
                                  schema = inputschema
                                 )
#display(inputfile)

from pyspark.sql.functions import col, when, lit
column_name = "testcol"
lstrip_udf = udf(lambda s: s.lstrip().lstrip("0"), StringType())
outputfile = (inputfile.withColumn(column_name,
                                  when(col(column_name).isNotNull(),
                                       lstrip_udf(col(column_name)) #replace this line with "Val123" and it works
                                      )
                                 ))
display(outputfile)

错误：

File "<command-3701821159856508>", line 18, in <lambda>
AttributeError: 'NoneType' object has no attribute 'lstrip'

谢谢

【参考方案1】：

这可能是 Spark 中的一个错误，所以这里对 UDF 做一个小的修改来解决这个问题：

lstrip_udf = udf(lambda s: s.lstrip().lstrip("0") if s is not None else None, StringType())

或者你可以使用 Spark SQL 来做这件事，比使用 UDF 更高效：

outputfile = (
    inputfile.withColumn(column_name,
        F.when(col(column_name).isNotNull(),
            F.expr("ltrim('0', ltrim('',testcol))")
        )
    )
)

【讨论】：

我认为你可能是对的@mck：issues.apache.org/jira/browse/SPARK-19844。看起来有人将票标记为“已解决”但没有发表评论，所以不清楚它是怎么回事：s

@user6386471 它说Labels: bulk-closed ;)

我也遇到过火花问题 19844，但由于它已“解决”，我认为问题已得到解决。因此，我的问题是，如果我在这里遗漏了一些概念性的东西，但如果我理解正确，这仍然是一个错误，并且上面链接的案例 19844 被设置为“已解决”而没有实际修复……在这种情况下，我很高兴我的理解是正确，这确实是一个错误，谢谢:)