如何检查DataFrame列中XML的良构性并解决UDF运行报错

Spark DataFrame是不可变数据结构，withColumn方法不会修改原DataFrame，而是返回一个新增了指定列的新DataFrame。你原有代码中没有接收withColumn的返回值，原df仍然只包含初始的三个列，所以后续过滤时会找不到Wellformed列。

from lxml import etree
from pyspark.sql import functions as F
from pyspark.sql.types import BooleanType

def wellformedness(xml):
    try:
        etree.fromstring(xml)
        return True
    except:
        return False

udf_wellformedness = F.udf(wellformedness, BooleanType())
# 接收withColumn返回的新DataFrame
df = df.withColumn('Wellformed', udf_wellformedness('MyColumn'))
df2 = df.filter(~df["Wellformed"])

普通Python UDF为行级处理，需要频繁在JVM和Python进程间传输数据，性能损耗较大，推荐以下两种优化方案：

• 使用原生Spark XML库：基于JVM实现的spark-xml库内置XML良构校验能力，性能比Python UDF高3~10倍，无需自定义解析逻辑。
• 改用向量化Pandas UDF：如果需要自定义校验逻辑，使用Pandas UDF做批处理，大幅降低进程间通信开销，实现代码如下：

import pandas as pd
from lxml import etree
from pyspark.sql import functions as F
from pyspark.sql.types import BooleanType

@F.pandas_udf(BooleanType())
def wellformedness(xml_series: pd.Series) -> pd.Series:
    return xml_series.apply(lambda x: etree.fromstring(x) is not None if pd.notna(x) else False)

df = df.withColumn('Wellformed', wellformedness('MyColumn'))
df2 = df.filter(~df.Wellformed)

内容的提问来源于stack exchange，提问作者Ronald Hensbergen