如何使用Scala与Spark处理符合指定Schema的XML文件？

嘿，我来帮你搞定这个XML转指定Spark Schema的需求！先把你的场景和需求理清楚：

你的目标Schema #

root
 |-- IdKey: long (nullable = true)
 |-- Value: string (nullable = true)
 |-- CDate: date (nullable = true)

待处理的XML结构（修正了示例里的标签笔误，应该是<IdKey>而非<IdKeyData>，如果实际是后者记得改代码哦） #

<Item>
 <CDate>2018-05-08T00:00:00</CDate>
 <ListItemData>
  <ItemData>
   <IdKey>2</IdKey>
   <Value>1</Value>
  </ItemData>
  <ItemData>
   <IdKey>61</IdKey>
   <Value>2</Value>
  </ItemData>
 </ListItemData>
</Item>

具体实现步骤 #

1. 先搞定依赖 #

首先得给你的Spark项目加上XML处理的依赖，用sbt的话就加这行（版本要匹配你的Spark版本，比如Spark 3.2.x对应0.15.0）：

libraryDependencies += "com.databricks" % "spark-xml_2.12" % "0.15.0"

2. 核心处理代码 #

直接上可运行的Scala代码，注释里写清楚每一步做什么：

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

object XmlProcessor {
  def main(args: Array[String]): Unit = {
    // 初始化SparkSession
    val spark = SparkSession.builder()
      .appName("XMLToTargetSchema")
      .master("local[*]") // 生产环境记得删掉这行，用集群模式
      .getOrCreate()

    // 定义你需要的目标Schema
    val targetSchema = StructType(Seq(
      StructField("IdKey", LongType, nullable = true),
      StructField("Value", StringType, nullable = true),
      StructField("CDate", DateType, nullable = true)
    ))

    // 读取XML文件，指定根标签和行标签
    val xmlRawDF = spark.read
      .format("com.databricks.spark.xml")
      .option("rootTag", "Item") // XML的根节点是<Item>
      .option("rowTag", "Item") // 每一条数据的标签也是<Item>
      .load("你的XML文件路径，比如./data/items.xml")

    // 关键步骤：展开嵌套的ItemData数组，把CDate关联到每一条记录，同时转换类型
    val finalDF = xmlRawDF
      // 把ListItemData下的ItemData数组拆成多行
      .select(col("CDate"), explode(col("ListItemData.ItemData")).alias("item"))
      // 提取字段并转换类型，匹配目标Schema
      .select(
        col("item.IdKey").cast(LongType),
        col("item.Value").cast(StringType),
        // 把XML里的带T的日期字符串转成Spark的Date类型
        to_date(col("CDate")).alias("CDate")
      )

    // 验证结果：打印Schema和数据
    println("最终Schema：")
    finalDF.printSchema()
    println("样例数据：")
    finalDF.show()

    spark.stop()
  }
}

3. 几个关键点说明 #

• explode函数：专门用来处理数组类型的字段，把一个数组拆成多行，这样每个ItemData就能单独成为一条记录
• 日期转换：用to_date把XML里的2018-05-08T00:00:00格式转成Spark的DateType，如果你的日期格式有特殊情况，可以指定格式，比如to_date(col("CDate"), "yyyy-MM-dd'T'HH:mm:ss")
• 标签匹配：如果你的XML里IdKey的标签真的是<IdKeyData>（示例里的笔误），记得把代码里的item.IdKey改成item.IdKeyData，不然会找不到字段

4. 额外小提示 #

• 一定要保证Spark XML依赖的版本和你的Spark版本兼容，比如Spark 3.3.x可以用0.16.0版本的spark-xml
• 如果你的XML文件有多个<Item>根节点，rowTag配置成Item就没问题，会自动识别每一个<Item>为一条数据
• 如果需要写入结果，直接用finalDF.write就行，比如写入Parquet或者CSV

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