(spark-xml) 使用 from_xml 函数解析 xml 列时仅接收 null

Question

我正在尝试使用 spark-xml 解析一个非常简单的 XML 字符串列，但我只能接收 null 值，即使正确填充了 XML。

我用来解析 xml 的 XSD 是：

<xs:schema attributeFormDefault="unqualified" elementFormDefault="qualified" xmlns:xs="http://www.w3.org/2001/XMLSchema">
    <xs:element name="note">
        <xs:complexType>
            <xs:sequence>
                <xs:element type="xs:string" name="from"/>
                <xs:element type="xs:string" name="to"/>
                <xs:element type="xs:string" name="message"/>
            </xs:sequence>
        </xs:complexType>
    </xs:element>
</xs:schema>

虽然列中以字符串形式显示的 XML 如下所示，但每个标签都已正确填充：

<?xml version="1.0" encoding="UTF-8"?>
<note>
    <from>Jani</from>
    <to>Tove</to>
    <message>Remember me this weekend</message>
</note>

我用 scala 编写的 spark 代码是这样的：

    // XML Schema
    val schema = XSDToSchema.read("<the XSD as string>")
    
    // Spark Structured Streaming (N.b. the column value contains the xml as string)
    import spark.implicits._
    var df = initSource(spark)
      .withColumn("parsed", from_xml($"value", schema, 
          Map(
              "mode" -> "FAILFAST",
              "nullValue"-> "",
              "rowTag" -> "note",
              "ignoreSurroundingSpaces" -> "true"
          )
      ))
      .select($"value",$"parsed.note.from", $"parsed.note.to", $"parsed.note.message")
      .writeStream
      .format("console")
      // .option("mode", "FAILFAST")
      // .option("nullValue", "")
      // .option("rowTag", "note")
      // .option("ignoreSurroundingSpaces","true")
      .outputMode("append")
      .start()
      .awaitTermination(30*1000)

打印此数据帧的架构（在 select 语句之前）将给出预期的架构

root
 |-- value: string (nullable = true)
 |-- parsed: struct (nullable = true)
 |    |-- note: struct (nullable = false)
 |    |    |-- from: string (nullable = false)
 |    |    |-- to: string (nullable = false)
 |    |    |-- message: string (nullable = false)

但是在控制台中打印结果时，我得到的只是null 值，如下所示：

....
-------------------------------------------
Batch: 0
-------------------------------------------
+--------------------+----+----+-------+
|               value|from|  to|message|
+--------------------+----+----+-------+
|<?xml version="1....|null|null|   null|
|<?xml version="1....|null|null|   null|
|<?xml version="1....|null|null|   null|
....

我认为这无关紧要，但此 xml 列的来源来自阅读定义如下的 Kafka 主题：

    def initSource(spark: SparkSession) : DataFrame = {
        spark.readStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "localhost:9092")
          .option("startingoffsets", "earliest")
          .option("subscribe", "my-test-topic")
          .load()
          .selectExpr("CAST(value AS STRING)")
    }

有其他人遇到过这个问题并解决了吗？我的选择已经不多了，我真的很感激这个提示:)

我使用的 spark-xml 版本是最新的 atm，0.12.0 和 spark 3.1.1。

更新

我在调用 writeStream 后错误地传递了 spark-xml 选项，而是需要将它们作为 from_xml 函数的第三个参数传递。我仍然只得到空值...

Answer 1

最后让我大开眼界的是阅读 spark-xml documentation 中提到的部分：

用于验证 XML 的 XSD 文件的路径每行单独

这意味着模式匹配是通过每一行而不是整个 XML 完成的，在这种情况下，我的示例的模式需要类似于以下内容：

val schema = StructType(Array(
        StructField("from", StringType, nullable = true),
        StructField("to", StringType, nullable = true),
        StructField("message", StringType, nullable = true)))

也可以使用 XSD 来完成：

<xs:schema attributeFormDefault="unqualified" elementFormDefault="qualified" xmlns:xs="http://www.w3.org/2001/XMLSchema">
    <xs:element type="xs:string" name="from"/>
    <xs:element type="xs:string" name="to"/>
    <xs:element type="xs:string" name="message"/>
</xs:schema>

这两种声明模式的方法对我有用。希望对以后的人有所帮助。