创建一个简单的自定义项

import re
from pyspark.sql.functions import udf

extract = udf(lambda s: list(map(lambda x: x.split('=')[1] if len(x)>0 else x,re.findall('(ProductName=\"\d+\")', s))), ArrayType(StringType()))

df.withColumn('values', extract('Product')).show(30,False)

+--------+--------------------+--------------------+
|Customer|             Product|              values|
+--------+--------------------+--------------------+
|       A|<XmlData ProductN...|     ["123", "1452"]|
|       B|<XmlData ProductN...|             ["123"]|
|       C|<XmlData ProductN...|["123", "1452", "...|
|       D|                    |                  []|
+--------+--------------------+--------------------+

df = df.withColumn('values',f.concat_ws(';','values'))

如果您的xml结构是静态的，只需将周围的xml字符串替换为“；”(使用replace或regexp\u replace），保留值，并使用结果创建一个新列（withcolumn方法）。

在斯卡拉-

val regexp_extractAll = udf((xml: String, exp: String) =>
      exp.r.findAllMatchIn(xml).map(_.group(1)).mkString(";")
    )

    df.withColumn("ProductName", regexp_extractAll($"Product", lit("""ProductName="(\d+)"""")))
      .show(false)

    /**
      * +--------+-----------------------------------------------------------------------------+--------+
      * |Customer|Product                                                                      |ProductName       |
      * +--------+-----------------------------------------------------------------------------+--------+
      * |A       |<XmlData ProductName="123">x</XmlData><XmlData ProductName="1452">y</XmlData>|123;1452|
      * |B       |<XmlData ProductName="123">x</XmlData>                                       |123     |
      * |C       |<XmlData ProductName="123">x</XmlData><XmlData ProductName="1452">y</XmlData>|123;1452|
      * +--------+-----------------------------------------------------------------------------+--------+
      */