Spark Scala通用方案：加载多row标签复杂XML到DataFrame并存为表

嘿，刚好踩过不少Spark处理复杂XML的坑，我来给你梳理一套通用方案，再针对你给的示例XML给出具体实现！

一、通用解决方案（适用于含多个row标签的复杂XML） #

1. 先搞定依赖 #

Spark本身没有内置XML数据源，得用Databricks开源的spark-xml库，根据你的Spark和Scala版本选对应版本：

• SBT依赖：

libraryDependencies += "com.databricks" % "spark-xml_2.12" % "0.15.0" // 示例版本，需匹配你的Spark版本

• Maven依赖：

<dependency>
  <groupId>com.databricks</groupId>
  <artifactId>spark-xml_2.12</artifactId>
  <version>0.15.0</version>
</dependency>

2. 核心配置与注意事项 #

• 指定行标签（rowTag）：如果XML有明确的行级元素（比如<book>），直接指定；如果有多个不同行标签（比如<book>、<magazine>），可以用*匹配所有子元素，或者后续用XPath过滤。
• 处理嵌套结构：用Spark的xpath_*系列函数（xpath_string、xpath_number等）提取嵌套元素；如果是数组型嵌套，用explode展开。
• 属性处理：XML中的属性会自动添加_前缀（比如<book id="0">中的id会变成_id），可以用alias重命名。
• Schema推断：开启inferSchema自动推断类型，或者手动定义Schema避免推断错误。

3. 通用步骤：加载XML → 处理 → 保存为表 #

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

// 初始化SparkSession
val spark = SparkSession.builder()
  .appName("ComplexXMLLoader")
  .master("local[*]") // 生产环境请移除
  .enableHiveSupport() // 如果要保存到Hive表，开启这个
  .getOrCreate()

import spark.implicits._

// 1. 加载XML文件
val rawDF = spark.read
  .format("com.databricks.spark.xml")
  .option("rowTag", "*") // 匹配根下所有行级元素，适合多row标签场景
  .option("inferSchema", "true")
  .option("treatEmptyValuesAsNulls", "true")
  .option("rootTag", "root") // 如果XML有统一根标签，指定它
  .load("path/to/your/xml/files")

// 2. 处理嵌套结构（示例：提取属性和嵌套元素）
val processedDF = rawDF.select(
  col("_id").alias("element_id"),
  xpath_string(col("."), "author").alias("author"), // XPath提取author元素
  xpath_number(col("price"), "price/price").alias("price"), // 提取嵌套的price数值
  col("_tag").alias("element_type") // 多row标签时，用_tag区分元素类型
)

// 3. 保存为表
// 方式1：临时视图（会话内可用）
processedDF.createOrReplaceTempView("xml_elements_temp")

// 方式2：Hive永久表
processedDF.write
  .mode("overwrite") // 可选：append/ignore/error
  .saveAsTable("default.xml_elements")

// 方式3：先存为Parquet再建外部表（推荐生产环境）
processedDF.write
  .mode("overwrite")
  .parquet("hdfs://path/to/parquet/xml_elements")

spark.sql("""
  CREATE EXTERNAL TABLE IF NOT EXISTS default.xml_elements_parquet
  USING PARQUET
  LOCATION 'hdfs://path/to/parquet/xml_elements'
""")

二、针对你的示例XML的具体实现 #

你的示例XML结构如下：

<book id="0">
  <author>Matthew</author>
  <publish_date>Sun Oct 01 00:00:00 EDT 2000</publish_date>
  <description>An in-depth look at creating applications with XML. </description>
  <price id = "1">
    <price>44.95</price>
    <genre>Computer</genre>
    <title>XML Developer's Guide</title>
  </price>
</book>

这里我们用XPath实现灵活提取任意层级的元素，比如同时加载<book>主信息和嵌套的<price>子元素：

代码实现 #

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

val spark = SparkSession.builder()
  .appName("BookXMLProcessing")
  .master("local[*]")
  .getOrCreate()

import spark.implicits._

// 1. 加载XML，指定<book>为行标签
val bookDF = spark.read
  .format("com.databricks.spark.xml")
  .option("rowTag", "book")
  .option("inferSchema", "true")
  .load("path/to/your/book.xml")

// 2. 用XPath提取任意元素（包括嵌套的price子元素）
val finalDF = bookDF.select(
  col("_id").alias("book_id"), // 提取book的id属性
  col("author"),
  col("publish_date"),
  col("description"),
  col("price._id").alias("price_id"), // 提取price的id属性
  xpath_string(col("price"), "price/price").alias("price_value"), // XPath提取嵌套的price数值
  xpath_string(col("price"), "price/genre").alias("genre"),
  xpath_string(col("price"), "price/title").alias("book_title")
)

// 查看结果
finalDF.show(false)

// 3. 保存为表
finalDF.write
  .mode("overwrite")
  .saveAsTable("default.books")

输出结果 #

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