Spark-XML处理大压缩XML内存溢出，求分布式解析生成规范化Parquet方案

问题背景 #

需处理压缩后181MB的bz2嵌套XML文件，目标是将Schema每个层级拆分为独立表，通过外键关联父行。当前遇到的核心问题：

• 未指定Schema时，自动推断阶段因内存不足崩溃
• 指定Schema后，因将整个XML解析为单一数据行，写入Parquet时单个Executor节点内存溢出，repartition无法生效（仅一行数据可拆分）

核心问题定位 #

你的读取配置将rowTag与rootTag均设为根节点riskAdjustmentClaimSelectionDetailReport，导致整个XML被解析成单一数据行，所有嵌套的数组、结构体数据全部集中在这一行中，直接引发单个节点内存过载。

解决方案步骤 #

1. 拆分XML读取逻辑，按子层级设置rowTag #

spark-xml库支持指定嵌套节点作为rowTag，实现并行分片解析。根据你的Schema结构，分阶段读取不同层级的数据：

读取根级元数据（含文件头）

提取根节点的基础字段，生成根级唯一标识作为后续关联的外键：

import org.apache.spark.sql.types.{StructType, StructField, StringType}
import org.apache.spark.sql.functions._

// 构造根级Schema（仅保留必要字段）
val rootSchema = StructType(Seq(
  StructField("_xmlns", StringType, true),
  StructField("includedFileHeader", schema.fields.find(_.name == "includedFileHeader").get.dataType.asInstanceOf[StructType], true)
))

val rootDF = sparkSession.read
  .schema(rootSchema)
  .format("xml")
  .option("rootTag", "riskAdjustmentClaimSelectionDetailReport")
  .option("rowTag", "riskAdjustmentClaimSelectionDetailReport")
  .xml(path)

// 添加根级唯一ID
val rootWithIdDF = rootDF.withColumn("root_id", monotonically_increasing_id())
rootWithIdDF.write.parquet("/path/to/root_table")

读取CalendarYear层级数据

指定includedCalendarYearCategory作为rowTag，并关联根级ID：

val calendarYearSchema = schema.fields.find(_.name == "includedCalendarYearCategory").get.dataType.asInstanceOf[StructType]

val calendarYearDF = sparkSession.read
  .schema(calendarYearSchema)
  .format("xml")
  .option("rootTag", "riskAdjustmentClaimSelectionDetailReport")
  .option("rowTag", "includedCalendarYearCategory")
  .xml(path)
  .withColumn("root_id", lit(rootWithIdDF.select("root_id").head.getLong(0)))
  .withColumn("calendar_year_id", monotonically_increasing_id())

calendarYearDF.write.parquet("/path/to/calendar_year_table")

读取PlanCategory层级数据

指定includedPlanCategory作为rowTag，关联CalendarYear层级ID：

val planCategorySchema = calendarYearSchema.fields.find(_.name == "includedPlanCategory").get.dataType.asInstanceOf[StructType]

val planCategoryDF = sparkSession.read
  .schema(planCategorySchema)
  .format("xml")
  .option("rootTag", "riskAdjustmentClaimSelectionDetailReport/includedCalendarYearCategory")
  .option("rowTag", "includedPlanCategory")
  .xml(path)
  .withColumn("calendar_year_id", lit(calendarYearDF.select("calendar_year_id").head.getLong(0)))
  .withColumn("plan_id", monotonically_increasing_id())

planCategoryDF.write.parquet("/path/to/plan_table")

2. 拆分数组类型子节点生成独立表 #

对于Schema中的数组字段（如includedMedicalClaimCategory），从父层级DataFrame中拆分出数组数据，生成独立表并关联父级ID：

// 拆分医疗理赔数据
val medicalClaimDF = planCategoryDF
  .select("plan_id", "includedMedicalClaimCategory.*")
  .withColumn("medical_claim_id", monotonically_increasing_id())

medicalClaimDF.write.parquet("/path/to/medical_claim_table")

// 拆分嵌套的补充记录数据
val supplementalRecordDF = medicalClaimDF
  .select("medical_claim_id", "includedSupplementalRecordCategory.*")
  .withColumn("supplemental_record_id", monotonically_increasing_id())

supplementalRecordDF.write.parquet("/path/to/supplemental_record_table")

3. 调整Spark内存配置优化性能 #

通过以下配置提升Executor处理大数据的能力：

# 提交Spark任务时的配置示例
spark-submit \
  --executor-memory 16G \
  --driver-memory 8G \
  --conf spark.memory.fraction=0.8 \
  --conf spark.sql.spill.enabled=true \
  your_application.jar

关键说明 #

• 必须避免将rowTag设为根节点，否则无法实现并行解析
• 每个层级添加自增ID作为外键，保证表间关联的唯一性
• 对于超大数组字段，拆分后写入Parquet可大幅降低单节点内存压力

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