(spark-xml)使用from_xml函数分析XML列时仅接收NULL [英] (spark-xml) Receiving only null when parsing xml column using from_xml function

问题描述

我试图使用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)

打印此dataFrame的架构(在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专栏的来源来自于阅读一个定义如下的卡夫卡主题：

    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带有spark3.1.1。

更新

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

推荐答案

最后让我睁开眼睛的是spark-xmldocumentation中提到的部分：

用于分别为每行验证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>

声明模式的这两种方法对我来说很管用。希望这对未来的某人有所帮助。

这篇关于(spark-xml)使用from_xml函数分析XML列时仅接收NULL的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！