使用Spark解析超大XML文件失败及无数据返回问题求助

你的场景回顾 #

你有这样结构的XML文件：

<?xml version="1.0" encoding="utf-8"?>
<SomeRoottag>
 <row Id="47513849" PostTypeId="1" />
 <row Id="4751323" PostTypeId="4" />
 <row Id="475546" PostTypeId="1" />
 <row Id="47597" PostTypeId="2" />
</SomeRoottag>

最初用这段代码解析10MB小文件正常，但处理50GB超大文件时因内存不足（仅20GB）失败：

df = sqlContext.read.format('xml').option("rowTag","SomeRoottag").load("/tmp/xmlfile.xml")
flat=df.withColumn("rows2",explode(df.row)).select("rows2.*")
flat.write.format("parquet").saveAsTable("xml_table")

后来尝试直接指定rowTag="row"却读取不到任何数据：

df = (sqlContext.read.format('xml').option("rowTag", "row").load("/tmp/someXML.xml"))
df.printSchema()
df.show()

输出为空Schema和空结果集。

问题根源与解决方案 #

1. 先解决「大文件OOM」的核心问题

你最初的写法错误地把根节点SomeRoottag设为rowTag，这会让Spark把整个XML文件当成单一行数据加载到内存，再通过explode拆分。50GB的文件全部塞进内存，自然触发内存溢出。

正确的做法是直接把重复的<row>节点设为rowTag，让Spark逐行解析每个row，而非加载整个文件：

# 正确读取方式：直接以row为解析单元
df = sqlContext.read.format('xml')\
    .option("rowTag", "row")\
    .load("/tmp/xmlfile.xml")

# 直接写入Hive表，无需额外explode
df.write.format("parquet").saveAsTable("xml_table")

这样Spark会用StAX流式解析每个row，每个分区只处理一部分数据，不会一次性加载整个文件到内存。

2. 解决「指定rowTag后无数据」的问题

你遇到的空结果问题，大概率是这几个原因：

• 文件路径或权限问题：确认/tmp/someXML.xml确实存在，且Spark有读取权限；分布式环境下要确保文件在HDFS上的路径正确。
• XML编码不匹配：如果文件不是utf-8编码，需手动指定编码参数：

  df = sqlContext.read.format('xml')\
      .option("rowTag", "row")\
      .option("encoding", "gbk")  # 替换为文件实际编码
      .load("/tmp/xmlfile.xml")
  

• 自动推断Schema失败：大文件的Schema自动推断可能出错，建议手动定义Schema强制解析：

  from pyspark.sql.types import StructType, StructField, IntegerType
  
  # 手动定义与XML属性匹配的Schema
  custom_schema = StructType([
      StructField("Id", IntegerType(), nullable=True),
      StructField("PostTypeId", IntegerType(), nullable=True)
  ])
  
  df = sqlContext.read.format('xml')\
      .option("rowTag", "row")\
      .schema(custom_schema)\
      .load("/tmp/xmlfile.xml")
  

• spark-xml版本兼容性：旧版本的com.databricks:spark-xml库可能对嵌套结构解析有bug，建议升级到0.15.0以上的稳定版。

3. 针对超大文件的额外优化

处理50GB级别的文件，还可以通过这些配置进一步规避内存压力：

• 调整Spark分区大小：修改spark.sql.files.maxPartitionBytes参数（默认128MB），比如设为1GB，让Spark把大文件拆成更多小分区处理：

  # 读取文件前设置
  sqlContext.setConf("spark.sql.files.maxPartitionBytes", "1073741824")  # 1GB
  

• 优化Executor内存配置：提交Spark任务时，给Executor分配足够内存，同时开启序列化优化：

  spark-submit \
      --executor-memory 16G \
      --driver-memory 8G \
      --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
      your_script.py
  

• 禁用Schema推断：手动指定Schema后，关闭自动推断可减少内存开销和解析时间：

  df = sqlContext.read.format('xml')\
      .option("rowTag", "row")\
      .option("inferSchema", "false")\
      .schema(custom_schema)\
      .load("/tmp/xmlfile.xml")
  

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