Spark3.0(Python)提取XML嵌套结构首个extension标签的方案

核心思路 #

放弃explode生成多行的方式，直接通过数组索引提取首个extension元素，再针对relatedItem做精准过滤提取。

具体实现步骤 #

假设读取XML后得到的DataFrame名为df，extension为数组类型字段，relatedItem是其下的子字段：

1. 提取首个extension
   用Spark 3.0支持的element_at函数直接取数组第一个元素（索引从1开始），或者用getItem(0)：

   from pyspark.sql.functions import element_at, col, filter
   
   # 方式一：element_at
   df = df.withColumn("first_extension", element_at(col("extension"), 1))
   # 方式二：getItem
   # df = df.withColumn("first_extension", col("extension").getItem(0))
   

2. 提取指定的relatedItem
   如果relatedItem是first_extension下的数组，可按属性筛选目标项，比如提取type为"reference"的条目：

   df = df.withColumn(
       "target_relatedItem",
       filter(
           col("first_extension.relatedItem"),
           lambda x: x.getField("type") == "reference"
       ).getItem(0)  # 取过滤后的第一个结果
   )
   

3. 清理冗余字段（可选）
   删除原始extension数组字段，精简结果：

   df = df.drop("extension")
   

完整示例代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import element_at, col, filter, when, size

# 初始化SparkSession
spark = SparkSession.builder.appName("XMLProcessing").getOrCreate()

# 读取XML文件（需提前引入spark-xml依赖）
df = spark.read.format("xml") \
    .option("rootTag", "root") \
    .option("rowTag", "record") \
    .load("path/to/your/xml/file.xml")

# 处理空数组情况，避免空指针
df = df.withColumn(
    "first_extension",
    when(size(col("extension")) > 0, element_at(col("extension"), 1)).otherwise(None)
)

# 提取指定类型的relatedItem
df = df.withColumn(
    "target_relatedItem",
    filter(
        col("first_extension.relatedItem"),
        lambda x: x["type"] == "reference"
    ).getItem(0)
)

# 查看结果
df.select("first_extension", "target_relatedItem").show(truncate=False)

注意事项 #

• 提交任务时需引入spark-xml包，比如--packages com.databricks:spark-xml_2.12:0.14.0（根据Scala版本调整）
• 如果relatedItem可能为空，同样可以用when+size组合做空值判断，避免报错

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