Spark DataFrame嵌套JSON父列添加与XML拆分技术问询

Got it, let's tackle your two Spark questions one by one with practical examples and explanations:

1. 如何在Spark DataFrame的嵌套JSON结构中，将父Schema的部分列添加到子节点中？ #

这个需求分两种常见场景，我分别给出代码示例和逻辑说明：

场景1：子节点是单个结构体（StructType） #

假设你的原始DataFrame来自这样的JSON数据：

{
  "user_id": 101,
  "user_name": "Alice",
  "profile": {
    "age": 30,
    "city": "New York"
  }
}

要把父列user_id嵌入到profile子结构体里，用struct函数重新构造子节点即可（Python代码）：

from pyspark.sql.functions import struct, col

# 假设原DataFrame名为df
df = df.withColumn(
    "profile",
    struct(
        col("profile.*"),  # 保留原profile的所有字段
        col("user_id")     # 把父列添加到子节点中
    )
)

处理后profile的结构会变成：

"profile": {
  "age": 30,
  "city": "New York",
  "user_id": 101
}

场景2：子节点是结构体数组（ArrayType(StructType)） #

如果嵌套结构是数组类型，比如订单数据：

{
  "order_id": 2001,
  "order_date": "2024-05-20",
  "items": [
    {"product_id": "P001", "quantity": 2},
    {"product_id": "P002", "quantity": 1}
  ]
}

要把父列order_id注入到items数组的每个元素中，需要用transform函数遍历数组：

from pyspark.sql.functions import transform, col

df = df.withColumn(
    "items",
    transform(
        col("items"),
        lambda item: struct(
            item["product_id"],
            item["quantity"],
            col("order_id")  # 将父列添加到数组的每个子元素里
        )
    )
)

处理后items数组会变成：

"items": [
  {"product_id": "P001", "quantity": 2, "order_id": 2001},
  {"product_id": "P002", "quantity": 1, "order_id": 2001}
]

2. 加载带命名空间的XML到Spark DataFrame，并生成两个目标DataFrame #

首先，你需要确保Spark环境依赖spark-xml扩展包，提交作业时可以通过以下命令引入：

spark-submit --packages com.databricks:spark-xml_2.12:0.15.0 your_script.py

完整实现流程 #

假设你的XML结构大致如下：

<env:ContentEnvelope xmlns:env="http://example.com/env" xmlns:fun="http://example.com/fun" xmlns:sr="http://example.com/sr">
  <env:Header>
    <!-- 头部无关内容 -->
  </env:Header>
  <env:Body>
    <fun:OrgId>ORG001</fun:OrgId>
    <fun:DataPartitionId>DP001</fun:DataPartitionId>
    <sr:Source>
      <sr:SourceId>SRC001</sr:SourceId>
      <sr:SourceName>OnlineStore</sr:SourceName>
    </sr:Source>
    <sr:Auditor>
      <sr:AuditId>AUD001</sr:AuditId>
      <sr:AuditDate>2024-05-20</sr:AuditDate>
      <sr:AuditorName>SystemAdmin</sr:AuditorName>
    </sr:Auditor>
  </env:Body>
</env:ContentEnvelope>

步骤1：加载XML并解析命名空间

from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

spark = SparkSession.builder.appName("XMLProcessing").getOrCreate()

# 加载XML，指定根节点、行节点和命名空间映射
xml_df = spark.read.format("xml") \
    .option("rootTag", "env:ContentEnvelope") \
    .option("rowTag", "env:Body") \
    .option("namespace", "env=http://example.com/env; fun=http://example.com/fun; sr=http://example.com/sr") \
    .load("path/to/your/xml/file.xml")

步骤2：提取全局固定值

因为fun:OrgId和fun:DataPartitionId对所有行值一致，直接提取一次即可：

org_id = xml_df.select("fun:OrgId").first()[0]
data_partition_id = xml_df.select("fun:DataPartitionId").first()[0]

步骤3：创建sr:Source对应的DataFrame

提取Source的所有字段，添加固定列：

source_df = xml_df.select("sr:Source.*") \
    .withColumn("OrgId", lit(org_id)) \
    .withColumn("DataPartitionId", lit(data_partition_id)) \
    .withColumn("action", lit("Overwrite"))

# 查看结果
source_df.show()

步骤4：补充sr:Auditor需求并创建对应DataFrame

我补充的合理需求是：提取Auditor下的核心审计字段（AuditId、AuditDate、AuditorName），关联全局固定的OrgId和DataPartitionId，同时添加固定action列；若Auditor存在嵌套子结构，可按需展开字段。实现代码：

auditor_df = xml_df.select("sr:Auditor.*") \
    .withColumn("OrgId", lit(org_id)) \
    .withColumn("DataPartitionId", lit(data_partition_id)) \
    .withColumn("action", lit("Overwrite"))

# 如果Auditor有嵌套结构（如<sr:AuditDetails>），可以这样展开：
# auditor_df = xml_df.select("sr:Auditor.*", "sr:Auditor.sr:AuditDetails.*") ...

# 查看结果
auditor_df.show()

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