Spark结构化流场景下如何从嵌套XML字符串提取目标字段

前置准备 #

需要在Spark应用的依赖中引入XML解析组件com.databricks:spark-xml，版本需匹配当前使用的Scala、Spark版本。你之前参考的公开教程实现失效，基本都是版本兼容问题、或原教程的Kafka直读XML场景和当前MQTT源拿字符串payload的场景逻辑不匹配导致的。

核心代码实现 #

基于你已有的MQTT流读取逻辑，无需调整流源部分代码，仅需在payload_字符串列基础上做转换即可：

1. 导入需要的Spark SQL函数与类型定义

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

2. 定义匹配目标XML结构的Schema，覆盖你需要提取的字段层级

val targetXmlSchema = StructType(Seq(
  StructField("ExecutionTime", StringType),
  StructField("FilterClass", StringType),
  StructField("InputData", StructType(Seq(
    StructField("FrdReq", StructType(Seq(
      StructField("HeaderSegment", StructType(Seq(
        StructField("Version", StringType),
        StructField("SequenceNb", StringType)
      )))
    )))
  )))
))

3. 配置XML解析适配参数，解决带命名空间的节点匹配问题

val xmlParseConfig = Map(
  "ignoreNamespace" -> "true",
  "rowTag" -> "CofiResults",
  "mode" -> "PERMISSIVE"
)

4. 逐行解析XML字符串并提取目标字段

val finalExtractDF = payload_.withColumn(
  "parsed_content",
  from_xml(col("payload"), targetXmlSchema, xmlParseConfig)
).select(
  col("parsed_content.ExecutionTime").as("execution_time"),
  col("parsed_content.FilterClass").as("filter_class"),
  col("parsed_content.InputData.FrdReq.HeaderSegment.Version").as("version"),
  col("parsed_content.InputData.FrdReq.HeaderSegment.SequenceNb").as("sequence_nb")
)

适配说明 #

• 该实现完全基于你现有的代码逻辑扩展，无需重写流读取部分
• 开启ignoreNamespace配置后，会自动忽略节点上的xmlns、ns2:类前缀，避免带命名空间的嵌套节点无法匹配Schema的问题
• 若需要提取节点上的属性值（比如示例中InputData节点的format、id属性），仅需在对应层级的Schema定义中补充属性字段，同时将解析配置中excludeAttribute设为false即可
• 解析模式设置为PERMISSIVE后，单条格式异常的XML不会导致整个流任务中断，异常内容会被标记为null值方便后续排查

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