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

Question

我在使用 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'

谢谢

Answer 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))")
        )
    )
)