在Databricks中将嵌套XML导入多关联表的技术需求问询

已完成步骤 #

1. XML导入DataFrame（带XSD验证） #

# Files
xml_file: str = "/FileStore/tables/demo.xml"
xsd_file: str = "/FileStore/tables/demo.xsd"

# Read XML file with XSD validation
df = spark.read \
    .format("xml") \
    .option("rowTag", "transactions") \
    .option("attributePrefix", "") \
    .option("rowValidationXSDPath", xsd_file) \
    .load(xml_file)

# Display the DataFrame
df.display()

2. 生成transactions表DataFrame #

from pyspark.sql.functions import monotonically_increasing_id, col

transactions_df = df.select(
    monotonically_increasing_id().alias("transaction_id"),
    "day",
    "art",
    "version",
    col("site.site_id").alias("site_id"),
    col("footer.tran_count").alias("tran_count")
)

transactions_df.display()

目标表结构说明 #

表1：transactions #

transaction_id：人工主键

表2：trans #

transaction_id：关联transactions.transaction_id的外键
trans_id：人工主键

表3：data_detail #

trans_id：关联trans.trans_id的外键
detail_id：人工主键

完整解决方案：生成trans和data_detail表DataFrame #

1. 生成trans表DataFrame #

展开嵌套的trans数组，关联主表主键并生成人工主键：

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, explode

# 展开trans数组，关联transaction_id
trans_expanded_df = transactions_df.select(
    "transaction_id",
    explode("trans").alias("trans")
)

# 提取trans中的嵌套字段
trans_df = trans_expanded_df.select(
    "transaction_id",
    col("trans.number").alias("number"),
    col("trans.meta.pro").alias("pro"),
    col("trans.meta.lib").alias("lib"),
    col("trans.meta.nr").alias("nr"),
    col("trans.details.header.at").alias("at"),
    col("trans.details.header.bt").alias("bt"),
    col("trans.details.header.ct").alias("ct"),
    col("trans.details.data.data_detail").alias("data_detail")  # 保留子节点用于生成data_detail表
)

# 按transaction_id分区排序，生成连续的trans_id
window_trans = Window.partitionBy("transaction_id").orderBy("number")
trans_df = trans_df.withColumn("trans_id", row_number().over(window_trans))

# 显示结果
trans_df.display()

2. 生成data_detail表DataFrame #

展开data_detail数组，关联trans表主键并生成全局唯一的人工主键：

# 展开data_detail数组，关联trans_id
data_detail_expanded_df = trans_df.select(
    "trans_id",
    explode("data_detail").alias("data_detail")
)

# 提取data_detail中的字段
data_detail_df = data_detail_expanded_df.select(
    "trans_id",
    col("data_detail.detail_id").alias("data_detail_id"),
    col("data_detail.typ").alias("type"),
    col("data_detail.class").alias("class")
)

# 全局排序生成连续的detail_id
window_detail = Window.orderBy("trans_id", "data_detail_id")
data_detail_df = data_detail_df.withColumn("detail_id", row_number().over(window_detail))

# 调整列顺序匹配目标表结构
data_detail_df = data_detail_df.select(
    "trans_id",
    "detail_id",
    "data_detail_id",
    "type",
    "class"
)

# 显示结果
data_detail_df.display()

表保存与关联查询验证 #

保存为Delta表 #

# 保存transactions表
transactions_df.write.format("delta").mode("overwrite").saveAsTable("default.transactions")

# 保存trans表（移除临时保留的data_detail列）
trans_final_df = trans_df.drop("data_detail")
trans_final_df.write.format("delta").mode("overwrite").saveAsTable("default.trans")

# 保存data_detail表
data_detail_df.write.format("delta").mode("overwrite").saveAsTable("default.data_detail")

关联查询示例 #

SELECT 
    t.transaction_id,
    t.art,
    tr.trans_id,
    tr.number,
    dd.detail_id,
    dd.type
FROM transactions t
JOIN trans tr ON t.transaction_id = tr.transaction_id
JOIN data_detail dd ON tr.trans_id = dd.trans_id
WHERE t.transaction_id = 1

附：XML与XSD文件内容 #

demo.xsd #

<?xml version="1.0" encoding="UTF-8"?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema" elementFormDefault="qualified">
  <xs:element name="transactions">
    <xs:complexType>
      <xs:sequence>
        <xs:element ref="site"/>
        <xs:element maxOccurs="unbounded" ref="trans"/>
        <xs:element ref="footer"/>
      </xs:sequence>
      <xs:attribute name="day" use="required" type="xs:integer"/>
      <xs:attribute name="art" use="required" type="xs:string"/>
      <xs:attribute name="version" use="required" type="xs:integer"/>
    </xs:complexType>
  </xs:element>
  <xs:element name="site">
    <xs:complexType>
      <xs:attribute name="site_id" use="required" type="xs:integer"/>
    </xs:complexType>
  </xs:element>
  <xs:element name="trans">
    <xs:complexType>
      <xs:all>
        <xs:element ref="number"/>
        <xs:element ref="meta"/>
        <xs:element ref="details"/>
      </xs:all>
    </xs:complexType>
  </xs:element>
  <xs:element name="footer">
    <xs:complexType>
      <xs:attribute name="tran_count" use="required" type="xs:integer"/>
    </xs:complexType>
  </xs:element>
  <xs:element name="number" type="xs:integer"/>
  <xs:element name="meta">
    <xs:complexType>
      <xs:sequence>
        <xs:element ref="pro"/>
        <xs:element ref="lib"/>
        <xs:element ref="nr"/>
      </xs:sequence>
    </xs:complexType>
  </xs:element>
  <xs:element name="pro" type="xs:string"/>
  <xs:element name="lib" type="xs:string"/>
  <xs:element name="nr" type="xs:string"/>
  <xs:element name="details">
    <xs:complexType>
      <xs:sequence>
        <xs:element ref="header"/>
        <xs:element ref="data"/>
      </xs:sequence>
    </xs:complexType>
  </xs:element>
  <xs:element name="header">
    <xs:complexType>
      <xs:sequence>
        <xs:element ref="at"/>
        <xs:element ref="bt"/>
        <xs:element ref="ct"/>
      </xs:sequence>
    </xs:complexType>
  </xs:element>
  <xs:element name="at" type="xs:integer"/>
  <xs:element name="bt" type="xs:integer"/>
  <xs:element name="ct" type="xs:integer"/>
  <xs:element name="data">
    <xs:complexType>
      <xs:sequence>
        <xs:element maxOccurs="unbounded" ref="data_detail"/>
      </xs:sequence>
    </xs:complexType>
  </xs:element>
  <xs:element name="data_detail">
    <xs:complexType>
      <xs:sequence>
        <xs:element ref="typ"/>
        <xs:element ref="class"/>
      </xs:sequence>
      <xs:attribute name="detail_id" use="required" type="xs:integer"/>
    </xs:complexType>
  </xs:element>
  <xs:element name="typ" type="xs:integer"/>
  <xs:element name="class" type="xs:integer"/>
</xs:schema>

demo.xml #

<?xml version="1.0" encoding="UTF-8"?>
<transactions day="1000" art="lo" version="1">
<site site_id="20"/>
<trans>
  <number>1234567</number>
  <meta>
    <pro>p1</pro>
    <lib>l1</lib>
    <nr>12345-1234567</nr>
  </meta>
  <details>
    <header>
      <at>123</at>
      <bt>1</bt>
      <ct>120</ct>
    </header>
    <data>
        <data_detail detail_id="0">
          <typ>5</typ>
          <class>10</class>
        </data_detail>
        <data_detail detail_id="1">
          <typ>6</typ>
          <class>11</class>
        </data_detail>
    </data>
  </details>
</trans>
<trans>
  <number>2345678</number>
  <meta>
    <pro>p2</pro>
    <lib>l2</lib>
    <nr>12345-2345678</nr>
  </meta>
  <details>
    <header>
      <at>456</at>
      <bt>1</bt>
      <ct>240</ct>
    </header>
    <data>
        <data_detail detail_id="0">
          <typ>7</typ>
          <class>20</class>
        </data_detail>
        <data_detail detail_id="1">
          <typ>8</typ>
          <class>21</class>
        </data_detail>
    </data>
  </details>
</trans>
<trans>
  <number>3456789</number>
  <meta>
    <pro>p3</pro>
    <lib>l3</lib>
    <nr>12345-3456789</nr>
  </meta>
  <details>
    <header>
      <at>789</at>
      <bt>1</bt>
      <ct>360</ct>
    </header>
    <data>
        <data_detail detail_id="0">
          <typ>9</typ>
          <class>30</class>
        </data_detail>
        <data_detail detail_id="1">
          <typ>10</typ>
          <class>31</class>
        </data_detail>
    </data>
  </details>
</trans>
<footer tran_count="3"/>
</transactions>

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