读取和写入 XML 文件

项目
02/10/2025

重要

此功能目前以公共预览版提供。

本文介绍如何读取和写入 XML 文件。

可扩展标记语言 (XML) 是一种标记语言，用于以文本格式设置数据格式、存储和共享数据。它定义了一 set 规则，用于序列化从文档到任意数据结构的数据。

本机 XML 文件格式支持启用对 XML 数据的引入、查询和分析，以便进行批处理或流式处理。它可以自动推断和改进架构和数据类型，支持 SQL 表达式（如 from_xml），并且可以生成 XML 文档。它不需要外部 jar，并且可与自动加载程序 read_files 和 COPY INTO 无缝配合工作。可以选择性地针对 XML Schema 定义（XSD）验证每个行级 XML 记录。

要求

Databricks Runtime 14.3 及更高版本

分析 XML 记录

XML 规范规定格式标准的结构。但是，此规范不会立即映射到表格格式。必须指定 rowTag 选项以指示映射到 DataFrameRow 的 XML 元素。该 rowTag 元素将成为顶级 struct 元素。 rowTag 的子元素将成为顶级 struct 的字段。

可以为此记录指定架构，也可以让它自动推断。由于分析程序仅检查 rowTag 元素，因此会筛选掉 DTD 和外部实体。

以下示例演示了使用不同 rowTag 选项对 XML 文件进行 schema 推断和解析：

Python

xmlString = """
  <books>
    <book id="bk103">
      <author>Corets, Eva</author>
      <title>Maeve Ascendant</title>
    </book>
    <book id="bk104">
      <author>Corets, Eva</author>
      <title>Oberon's Legacy</title>
    </book>
  </books>"""

xmlPath = "dbfs:/tmp/books.xml"
dbutils.fs.put(xmlPath, xmlString, True)

Scala

val xmlString = """
  <books>
    <book id="bk103">
      <author>Corets, Eva</author>
      <title>Maeve Ascendant</title>
    </book>
    <book id="bk104">
      <author>Corets, Eva</author>
      <title>Oberon's Legacy</title>
    </book>
  </books>"""
val xmlPath = "dbfs:/tmp/books.xml"
dbutils.fs.put(xmlPath, xmlString)

读取 rowTag 选项为“books”的 XML 文件：

Python

df = spark.read.option("rowTag", "books").format("xml").load(xmlPath)
df.printSchema()
df.show(truncate=False)

Scala

val df = spark.read.option("rowTag", "books").xml(xmlPath)
df.printSchema()
df.show(truncate=false)

输出：

root
|-- book: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- _id: string (nullable = true)
| | |-- author: string (nullable = true)
| | |-- title: string (nullable = true)

+------------------------------------------------------------------------------+
|book                                                                          |
+------------------------------------------------------------------------------+
|[{bk103, Corets, Eva, Maeve Ascendant}, {bk104, Corets, Eva, Oberon's Legacy}]|
+------------------------------------------------------------------------------+

读取 rowTag 为“book”的 XML 文件：

Python

df = spark.read.option("rowTag", "book").format("xml").load(xmlPath)
# Infers three top-level fields and parses `book` in separate rows:

Scala

val df = spark.read.option("rowTag", "book").xml(xmlPath)
// Infers three top-level fields and parses `book` in separate rows:

输出：

root
|-- _id: string (nullable = true)
|-- author: string (nullable = true)
|-- title: string (nullable = true)

+-----+-----------+---------------+
|_id  |author     |title          |
+-----+-----------+---------------+
|bk103|Corets, Eva|Maeve Ascendant|
|bk104|Corets, Eva|Oberon's Legacy|
+-----+-----------+---------------+

数据源选项

可以通过以下方式指定 XML 的数据源选项：

以下 .option/.options 方法：
- DataFrameReader
- DataFrameWriter
- DataStreamReader
- DataStreamWriter
以下内置函数：
CREATE TABLE USING DATA_SOURCE 的 OPTIONS 字句

如需选项列表，请参阅自动加载程序选项。

XSD 支持

可以选择性地通过 XML Schema 定义（XSD）来验证每个行级 XML 记录。 XSD 文件在 rowValidationXSDPath 选项中指定。 XSD 不会以其他方式影响提供或推理的架构。验证失败的记录将标记为“已损坏”，并将根据选项部分中所述的损坏记录处理模式选项进行处理。

可以使用 XSDToSchema 从 XSD 文件提取 Spark 数据帧 schema。它仅支持简单类型、复杂类型和序列类型，仅支持基本 XSD 功能。

import org.apache.spark.sql.execution.datasources.xml.XSDToSchema
import org.apache.hadoop.fs.Path

val xsdPath = "dbfs:/tmp/books.xsd"
val xsdString = """<?xml version="1.0" encoding="UTF-8" ?>
  <xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
    <xs:element name="book">
      <xs:complexType>
        <xs:sequence>
          <xs:element name="author" type="xs:string" />
          <xs:element name="title" type="xs:string" />
          <xs:element name="genre" type="xs:string" />
          <xs:element name="price" type="xs:decimal" />
          <xs:element name="publish_date" type="xs:date" />
          <xs:element name="description" type="xs:string" />
        </xs:sequence>
        <xs:attribute name="id" type="xs:string" use="required" />
      </xs:complexType>
    </xs:element>
  </xs:schema>"""

dbutils.fs.put(xsdPath, xsdString, true)

val schema1 = XSDToSchema.read(xsdString)
val schema2 = XSDToSchema.read(new Path(xsdPath))

以下 table 显示了 XSD 数据类型到 Spark 数据类型的转换：

XSD 数据类型	Spark 数据类型
`boolean`	`BooleanType`
`decimal`	`DecimalType`
`unsignedLong`	`DecimalType(38, 0)`
`double`	`DoubleType`
`float`	`FloatType`
`byte`	`ByteType`
%>	`ShortType`
`integer`, `negativeInteger`, `nonNegativeInteger`, `nonPositiveInteger`, `positiveInteger`, `unsignedShort`	`IntegerType`
%>	`LongType`
`date`	`DateType`
`dateTime`	`TimestampType`
`Others`	`StringType`

分析嵌套 XML

可以使用 schema_of_xml 和 from_xml 分析现有 DataFrame 的字符串值列中的 XML 数据，并将架构和分析的结果作为新 struct 列返回。作为自变量传递到 schema_of_xml 和 from_xml 的 XML 数据必须是单个格式标准的 XML 记录。

schema_of_xml

语法

schema_of_xml(xmlStr [, options] )

参数

xmlStr：一个 STRING 表达式，用于指定单个格式标准的 XML 记录。
options：指定指令的可选MAP<STRING,STRING>文本。

一个包含结构定义的字符串，其中有 n 个字符串字段，where 和 column 的名称是从 XML 元素和属性名称派生的。字段 values 保存派生格式的 SQL 类型。

from_xml

语法

from_xml(xmlStr, schema [, options])

参数

xmlStr：一个 STRING 表达式，用于指定单个格式标准的 XML 记录。
schema：schema_of_xml 函数的 STRING 表达式或调用。
options：指定指令的可选MAP<STRING,STRING>文本。

具有与 schema 定义匹配的字段名称和类型的结构。架构必须定义为逗号分隔的列名称和数据类型对（例如，与在 CREATE TABLE 中使用一样）。数据源选项中显示的大多数选项都适用，但有以下例外情况：

rowTag：由于只有一条 XML 记录，因此 rowTag 选项不适用。
mode（默认值为 PERMISSIVE）：允许采用在分析期间处理损坏记录的模式。
- PERMISSIVE：遇到损坏的记录时，将格式错误的字符串放入由 columnNameOfCorruptRecord 配置的字段中，并将格式错误的字段设置为 null。若要保留损坏的记录，可以在用户定义的架构中设置名为 columnNameOfCorruptRecord 的字符串类型字段。如果 schema 没有该字段，则会在分析过程中删除损坏的记录。在推断 schema时，它会在输出 schema隐式添加 columnNameOfCorruptRecord 字段。
- FAILFAST：遇到损坏的记录时引发异常。

结构转换

由于 DataFrame 和 XML 之间存在结构差异，因此对于从 XML 数据转换为 DataFrame 以及从 DataFrame 转换为 XML 数据来说，有一些转换规则。请注意，可以使用选项 excludeAttribute 禁用处理属性。

从 XML 转换为 DataFrame

属性：属性将转换为具有标题前缀 attributePrefix 的字段。

<one myOneAttrib="AAAA">
  <two>two</two>
  <three>three</three>
</one>

生成以下架构：

root
|-- _myOneAttrib: string (nullable = true)
|-- two: string (nullable = true)
|-- three: string (nullable = true)

包含属性或子元素的元素中的字符数据：这些数据在 valueTag 字段中解析。如果字符数据多次出现，则 valueTag 字段将转换为类型 array。

<one>
  <two myTwoAttrib="BBBBB">two</two>
  some value between elements
  <three>three</three>
  some other value between elements
</one>

生成以下架构：

root
 |-- _VALUE: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- two: struct (nullable = true)
 |    |-- _VALUE: string (nullable = true)
| |-- _myTwoAttrib: string (nullable = true)
|-- three: string (nullable = true)

从 DataFrame 转换为 XML

元素作为数组中的数组：从具有 ArrayType 字段且其元素为 ArrayType 的 DataFrame 写入 XML 文件时，将为该元素提供额外的嵌套字段。读写 XML 数据时不会发生这种情况，但写入从其他源读取的 DataFrame 时会。因此，读写和写读 XML 文件具有同一结构，但写入从其他源读取的 DataFrame 可能具有另一结构。

DataFrame 具有以下架构：

|-- a: array (nullable = true)
| |-- element: array (containsNull = true)
| | |-- element: string (containsNull = true)

以及以下数据：

+------------------------------------+
| a|
+------------------------------------+
|[WrappedArray(aa), WrappedArray(bb)]|
+------------------------------------+

生成以下 XML 文件：

<a>
  <item>aa</item>
</a>
<a>
  <item>bb</item>
</a>

选项 DataFrame 指定 arrayElementName 中未命名数组的元素名称（默认值：item）。

补救数据列

已获救的数据 column 可确保在 ETL 期间永远不会丢失或错过数据。可以启用已获救数据 column 捕获未分析的任何数据，因为记录中的一个或多个字段存在下列问题之一：

不存在于提供的架构中
与提供的 schema 的数据类型不匹配
与提供的架构中的字段名称大小写不匹配

已获救的数据 column 作为 JSON 文档返回，其中包含已获救的 columns，以及记录的源文件路径。若要从补救的数据列中删除源文件路径，可以设置以下 SQL 配置：

Python

spark.conf.set("spark.databricks.sql.rescuedDataColumn.filePath.enabled", "false")

Scala

spark.conf.set("spark.databricks.sql.rescuedDataColumn.filePath.enabled", "false").

你可以通过在读取数据时将选项 rescuedDataColumn 设置为某个列名来启用补救的数据列，例如，spark.read.option("rescuedDataColumn", "_rescued_data").format("xml").load(<path>) 的 _rescued_data。

分析记录时，XML 分析程序支持三种模式：PERMISSIVE、DROPMALFORMED 和 FAILFAST。与 rescuedDataColumn 一起使用时，数据类型不匹配不会导致在 DROPMALFORMED 模式下删除记录，或者在 FAILFAST 模式下引发错误。只有损坏的记录（即不完整或格式错误的 XML）会被删除或引发错误。

自动加载程序中的架构参考和演化

有关本主题和适用选项的详细讨论，请参阅在自动加载程序中配置 schema 推理和演变。可以将自动加载程序配置为自动检测已加载的 XML 数据的 schema，从而无需显式声明数据 schema 即可初始化 tables，并在引入新 columns 时改进 tableschema。这样就无需一直手动跟踪和应用架构更改。

默认情况下，Auto Loader schema 推理旨在避免因类型不匹配而导致的 schema 演变问题。对于不编码数据类型（JSON、CSV 和 XML）的格式，自动加载程序将所有 columns 推断为字符串，包括 XML 文件中的嵌套字段。 Apache Spark DataFrameReader 使用不同的行为进行 schema 推理，根据示例数据为 XML 源中的 columns 选择数据类型。若要使用自动加载程序实现此行为，请将选项 cloudFiles.inferColumnTypes 设置为 true。

自动加载程序在处理数据时会检测是否添加了新列。当自动加载程序检测到新的 column时，流会停止并显示 UnknownFieldException。在流引发此错误之前，自动加载程序会在最新的数据微批上执行架构推理，并通过将新列合并到架构末尾来使用最新架构更新架构位置。现有 columns 的数据类型保持不变。自动加载程序支持不同的架构演变模式，可以在选项 cloudFiles.schemaEvolutionMode 中设置这些模式。

通过使用架构提示，可以强制实施你所知道的并期望出现在推理架构中的架构信息。如果你知道某列采用特定的属性类型，或者想要选择更常规的数据类型（例如不使用整数，而改用双精度），可以为列数据类型提供任意数量的提示（格式为符合 SQL 架构规范语法的字符串）。启用补救数据列时，以非架构所用的大小写形式命名的字段将加载到 _rescued_data 列。你可以通过将选项 readerCaseSensitive 设置为 false 来更改此行为，在这种情况下自动加载程序将以不区分大小写的方式读取数据。

示例

本部分中的示例使用可在 Apache Spark GitHub 存储库中下载的 XML 文件。