Чтение и запись XML-файлов

Внимание

Эта функция предоставляется в режиме общедоступной предварительной версии.

Расширяемый язык разметки (XML) — это язык разметки для форматирования, хранения и совместного использования данных в текстовом формате. Он определяет набор правил сериализации данных, начиная от документов до произвольных структур данных.

Azure Databricks поддерживает XML для чтения и записи с помощью Apache Spark, включая автоматическое вывод схемы и эволюцию, конфигурацию тегов строк, проверку XSD и выражения SQL, такие какfrom_xml. Встроенная поддержка XML работает с Auto Loader, read_files и COPY INTO без необходимости использовать внешние JAR-файлы.

Prerequisites

Для поддержки формата XML-файла требуется Databricks Runtime 14.3 и более поздней версии.

Options

Используйте методы .option() и .options()DataFrameReader и DataFrameWriter для настройки источников данных XML. Полный список поддерживаемых параметров см. в разделе DataFrameReader "Параметры XML " и DataFrameWriter "ПАРАМЕТРЫ XML".

Анализ XML-записей

Спецификация XML требует хорошо сформированной структуры. Однако эта спецификация не сразу может быть представлена в табличном формате. Необходимо указать rowTag параметр, указывающий XML-элемент, который сопоставляется с элементом DataFrameRow. Элемент rowTag становится верхним уровнем struct. Дочерние элементы rowTag становятся полями верхнего уровня struct.

Можно указать схему для этой записи или позволить ей автоматически выводить. Поскольку анализатор проверяет только элементы rowTag, DTD и внешние сущности отфильтровываются.

В следующих примерах показано вывод схемы и анализ XML-файла с помощью различных rowTag параметров:

Питон

xmlString = """
  <reviews>
    <review id="r001">
      <author>Alice</author>
      <rating>5</rating>
      <comment>Amazing stay, highly recommend!</comment>
    </review>
    <review id="r002">
      <author>Bob</author>
      <rating>4</rating>
      <comment>Great location, very comfortable</comment>
    </review>
  </reviews>"""

xmlPath = "/Volumes/<catalog>/<schema>/<volume>/reviews.xml"
dbutils.fs.put(xmlPath, xmlString, True)

язык программирования Scala

val xmlString = """
  <reviews>
    <review id="r001">
      <author>Alice</author>
      <rating>5</rating>
      <comment>Amazing stay, highly recommend!</comment>
    </review>
    <review id="r002">
      <author>Bob</author>
      <rating>4</rating>
      <comment>Great location, very comfortable</comment>
    </review>
  </reviews>"""
val xmlPath = "/Volumes/<catalog>/<schema>/<volume>/reviews.xml"
dbutils.fs.put(xmlPath, xmlString)

Прочитайте XML-файл с параметром rowTag как "reviews":

Питон

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

язык программирования Scala

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

SQL

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews.xml',
  format => 'xml',
  rowTag => 'reviews'
)

Выходные данные:

root
|-- review: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- _id: string (nullable = true)
| | |-- author: string (nullable = true)
| | |-- comment: string (nullable = true)
| | |-- rating: string (nullable = true)

+----------------------------------------------------------------------------------------+
|review                                                                                  |
+----------------------------------------------------------------------------------------+
|[{r001, Alice, Amazing stay, highly recommend!, 5}, {r002, Bob, Great location..., 4}] |
+----------------------------------------------------------------------------------------+

Прочитайте XML-файл с помощью rowTag как "review":

Питон

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

язык программирования Scala

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

SQL

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews.xml',
  format => 'xml',
  rowTag => 'review'
)

Выходные данные:

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

+----+------+--------------------------------+------+
|_id |author|comment                         |rating|
+----+------+--------------------------------+------+
|r001|Alice |Amazing stay, highly recommend! |5     |
|r002|Bob   |Great location, very comfortable|4     |
+----+------+--------------------------------+------+

Проверка XML-записей с помощью XSD

При необходимости можно проверить каждую xml-запись уровня строки с помощью определения схемы XML (XSD). XSD-файл указывается в параметре rowValidationXSDPath . В противном случае XSD не влияет на указанную или логически выведенную схему. Запись, которая не проходит проверку, помечается как "поврежденная" и обрабатывается в соответствии с параметром режима обработки поврежденных записей, описанным в разделе параметров.

Можно использовать XSDToSchema для извлечения схемы Кадра данных Spark из XSD-файла. Он поддерживает только простые, сложные и последовательные типы и поддерживает только основные функции XSD.

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

val xsdPath = "/Volumes/<catalog>/<schema>/<volume>/reviews.xsd"
val xsdString = """<?xml version="1.0" encoding="UTF-8" ?>
  <xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
    <xs:element name="review">
      <xs:complexType>
        <xs:sequence>
          <xs:element name="author" type="xs:string" />
          <xs:element name="rating" type="xs:integer" />
          <xs:element name="comment" 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))

В следующей таблице показано преобразование типов данных XSD в типы данных Spark:

Типы данных XSD Типы данных Spark
boolean BooleanType
decimal DecimalType
unsignedLong DecimalType(38, 0)
double DoubleType
float FloatType
byte ByteType
short, unsignedByte ShortType
integer, negativeInteger, nonNegativeInteger, nonPositiveInteger, positiveInteger, unsignedShort IntegerType
long, unsignedInt LongType
date DateType
dateTime TimestampType
Others StringType

Анализ вложенных XML-файлов

XML-данные в строковом столбце существующего DataFrame можно разобрать с помощью schema_of_xml и from_xml, которые возвращают схему и результаты разбора в виде новых столбцов struct. XML-данные, передаваемые в качестве аргумента schema_of_xml и from_xml должны быть одной хорошо сформированной XML-записью.

схема XML

Используется schema_of_xml для вывода схемы Spark из XML-строки. Передайте результат from_xml для анализа XML-столбцов.

Синтаксис: schema_of_xml(xmlStr [, options])

Аргумент Required Description
xmlStr Yes Выражение STRING, указывающее одну хорошо сформированную XML-запись.
options Нет Литерал MAP<STRING,STRING> , указывающий директивы.

Возвращает строку, содержащую определение структуры с n полями строк, в которых имена столбцов являются производными от XML-элемента и имен атрибутов. Значения полей содержат производные отформатированные типы SQL.

from_xml

Используйте from_xml для разбора столбца STRING, содержащего XML-записи, в структуру (struct). Укажите схему напрямую или используйте выходные данные schema_of_xml.

Синтаксис: from_xml(xmlStr, schema [, options])

Аргумент Required Description
xmlStr Yes Выражение STRING, указывающее одну хорошо сформированную XML-запись.
schema Yes Выражение STRING или вызов функции schema_of_xml.
options Нет Литерал MAP<STRING,STRING> , указывающий директивы.

Возвращает структуру с именами полей и типами, соответствующими определению схемы. Схема должна быть определена как разделённые запятыми пары имени столбца и типа данных, как, например, в CREATE TABLE. Большинство параметров, показанных в разделе "Параметры" , применимы со следующими исключениями:

  • rowTag: Так как имеется только одна XML-запись, параметр rowTag неприменим.
  • mode (по умолчанию: PERMISSIVE: разрешает режим для работы с поврежденными записями во время синтаксического анализа.
    • PERMISSIVE: при обнаружении поврежденной записи помещает неправильно сформированную строку в поле, заданное параметром columnNameOfCorruptRecord, а неправильно сформированным полям присваивает значение null. Чтобы сохранить поврежденные записи, можно задать поле типа строки с именем columnNameOfCorruptRecord в определяемой пользователем схеме. Если в схеме нет этого поля, то поврежденные записи удаляются во время синтаксического анализа. При выводе схемы поле columnNameOfCorruptRecord неявным образом добавляется в выходную схему.
    • FAILFAST: вызывает исключение при обнаружении поврежденных записей.

Примеры

Чтобы проанализировать столбец строки XML, используйте schema_of_xml для вывода схемы, а затем передайте ее в from_xml:

Питон

from pyspark.sql.functions import from_xml, schema_of_xml, lit, col

xml_data = """
  <review id="r001">
    <author>Alice</author>
    <rating>5</rating>
    <comment>Amazing stay, highly recommend!</comment>
  </review>
"""

df = spark.createDataFrame([(1, xml_data)], ["review_id", "payload"])
schema = schema_of_xml(df.select("payload").limit(1).collect()[0][0])
parsed = df.withColumn("parsed", from_xml(col("payload"), schema))
parsed.printSchema()
parsed.show()

язык программирования Scala

import org.apache.spark.sql.functions.{from_xml, schema_of_xml, lit}

val xmlData = """
  <review id="r001">
    <author>Alice</author>
    <rating>5</rating>
    <comment>Amazing stay, highly recommend!</comment>
  </review>""".stripMargin

val df = Seq((1, xmlData)).toDF("review_id", "payload")
val schema = schema_of_xml(xmlData)
val parsed = df.withColumn("parsed", from_xml($"payload", schema))
parsed.printSchema()
parsed.show()

Чтобы проанализировать встроенный XML-код в SQL, выполните приведенные далее действия.

SELECT from_xml('
  <review id="r001">
    <author>Alice</author>
    <rating>5</rating>
    <comment>Amazing stay, highly recommend!</comment>
  </review>',
  schema_of_xml('
  <review id="r001">
    <author>Alice</author>
    <rating>5</rating>
    <comment>Amazing stay, highly recommend!</comment>
  </review>')
);

Преобразовать структуры XML в DataFrame и обратно

Из-за различий в структуре между DataFrame и XML существуют правила преобразования XML-данных в DataFrame и из DataFrame в XML-данные. Обратите внимание, что обработка атрибутов может быть отключена с помощью параметра excludeAttribute.

Преобразование XML в DataFrame

При чтении XML Azure Databricks сопоставляет 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

При записи DataFrame в XML некоторые вложенные структуры требуют особой обработки из-за различий между моделями данных DataFrame и XML.

Если DataFrame содержит поле ArrayType, тип элементов которого также является ArrayType, то при записи его в XML создается дополнительный уровень вложенности, которого нет при чтении и последующей записи XML-файлов. Это влияет только на таблицы DataFrame, полученные не из XML, — чтение и запись XML-файлов сохраняют исходную структуру.

Например, 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).

Включить столбец восстановленных данных

Столбец спасенных данных гарантирует, что данные во время ETL никогда не теряются. Он записывает данные, которые не были проанализированы, так как в одной или нескольких полях записи возникает одна из следующих проблем:

  • Отсутствует в предоставленной схеме.
  • Не соответствует типу данных предоставленной схемы.
  • Имеется несовпадение регистра с названиями полей в предоставленной схеме.

Спасательный столбец данных возвращается в виде документа JSON, содержащего столбцы, которые были спасены, и путь к исходному файлу записи.

Чтобы включить резервный столбец данных, задайте для параметра rescuedDataColumn имя столбца при чтении:

Питон

df = spark.read.option("rescuedDataColumn", "_rescued_data").format("xml").load("/Volumes/<catalog>/<schema>/<volume>/reviews_xml")

язык программирования Scala

val df = spark.read.option("rescuedDataColumn", "_rescued_data").format("xml").load("/Volumes/<catalog>/<schema>/<volume>/reviews_xml")

SQL

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews_xml',
  format => 'xml',
  rowTag => 'review',
  rescuedDataColumn => '_rescued_data'
)

Чтобы удалить путь к исходному файлу из спасаемого столбца данных, установите следующий параметр:

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

Средство синтаксического анализа XML поддерживает три режима при анализе записей: PERMISSIVE, DROPMALFORMEDи FAILFAST. При использовании вместе с rescuedDataColumn несоответствие типов данных не приводит к удалению записей в режиме DROPMALFORMED или возникновению ошибки в режиме FAILFAST. Только поврежденные записи (неполный или некорректный XML) отбрасываются или вызывают ошибки.

Определение и эволюция схемы с помощью Auto Loader

Подробные сведения об этом разделе и применимых параметрах см. в разделе "Настройка вывода схемы и эволюции" в автозагрузчике. Вы можете настроить Auto Loader на автоматическое определение схемы загружаемых XML-данных, что позволяет инициализировать таблицы без явного объявления схемы данных и изменять схему таблицы по мере добавления новых столбцов. Это устраняет необходимость вручную отслеживать и применять изменения схемы с течением времени.

По умолчанию определение схемы в Auto Loader направлено на предотвращение проблем эволюции схемы из-за несоответствия типов. Для форматов, которые не кодируют типы данных (JSON, CSV и XML), автозагрузчик выводит все столбцы в виде строк, включая вложенные поля в XML-файлах. Apache Spark DataFrameReader использует другое поведение для вывода схемы, выбирая типы данных для столбцов в XML-источниках на основе примеров данных. Чтобы включить это поведение с помощью автозагрузчика, задайте для параметра значение cloudFiles.inferColumnTypestrue.

Автозагрузчик обнаруживает добавление новых столбцов при обработке данных. Когда Auto Loader обнаруживает новый столбец, поток останавливается с ошибкой UnknownFieldException. Прежде чем поток выдает эту ошибку, автозагрузчик выполняет вывод схемы в последней микропакете данных и обновляет расположение схемы с последней схемой, объединяя новые столбцы в конец схемы. Типы данных существующих столбцов останутся неизменными. Автозагрузчик поддерживает различные режимы эволюции схемы, заданные в параметре cloudFiles.schemaEvolutionMode.

Вы можете использовать подсказки для схемы, чтобы принудительно задать информацию о схеме, которая вам известна и которую вы ожидаете в автоматически выведенной схеме. Если вы знаете, что столбец имеет определенный тип данных или вы хотите выбрать более общий тип данных (например, двойной вместо целого числа), можно указать произвольное количество подсказок для типов данных столбцов в виде строки с помощью синтаксиса спецификации схемы SQL. При включении спасаемого столбца данных поля, именованные в случае, отличном от схемы, загружаются в _rescued_data столбец. Это поведение можно изменить, установив для параметра readerCaseSensitive значение false; в этом случае Auto Loader считывает данные без учета регистра.

Usage

В следующих примерах используется набор данных Wanderbricks для демонстрации чтения и записи XML-файлов с помощью API Spark DataFrame и SQL.

Чтение и запись XML

Используйте API Кадра данных для записи отзывов Wanderbricks в XML и их обратного чтения.

Питон

# Write Wanderbricks reviews to XML
df = spark.read.table("samples.wanderbricks.reviews")
df.write \
  .format("xml") \
  .option("rootTag", "reviews") \
  .option("rowTag", "review") \
  .save("/Volumes/<catalog>/<schema>/<volume>/reviews.xml")

# Read the XML file back
df_read = spark.read \
  .format("xml") \
  .option("rowTag", "review") \
  .load("/Volumes/<catalog>/<schema>/<volume>/reviews.xml")
df_read.show()

язык программирования Scala

// Write Wanderbricks reviews to XML
val df = spark.read.table("samples.wanderbricks.reviews")
df.write
  .format("xml")
  .option("rootTag", "reviews")
  .option("rowTag", "review")
  .save("/Volumes/<catalog>/<schema>/<volume>/reviews.xml")

// Read the XML file back
val dfRead = spark.read
  .format("xml")
  .option("rowTag", "review")
  .xml("/Volumes/<catalog>/<schema>/<volume>/reviews.xml")
dfRead.show()

Р

df <- loadDF("/Volumes/<catalog>/<schema>/<volume>/reviews.xml", source = "xml", rowTag = "review")
saveDF(df, "/Volumes/<catalog>/<schema>/<volume>/newreviews.xml", "xml", "overwrite")

Можно вручную указать схему при чтении данных:

Питон

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

custom_schema = StructType([
    StructField("_id", StringType(), True),
    StructField("author", StringType(), True),
    StructField("rating", IntegerType(), True),
    StructField("comment", StringType(), True)
])
df = spark.read.options(rowTag='review').xml('/Volumes/<catalog>/<schema>/<volume>/reviews.xml', schema=custom_schema)
df.show()

язык программирования Scala

import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}

val customSchema = StructType(Array(
  StructField("_id", StringType, nullable = true),
  StructField("author", StringType, nullable = true),
  StructField("rating", IntegerType, nullable = true),
  StructField("comment", StringType, nullable = true)))
val df = spark.read.option("rowTag", "review").schema(customSchema).xml("/Volumes/<catalog>/<schema>/<volume>/reviews.xml")
df.show()

Р

customSchema <- structType(
  structField("_id", "string"),
  structField("author", "string"),
  structField("rating", "integer"),
  structField("comment", "string"))

df <- loadDF("/Volumes/<catalog>/<schema>/<volume>/reviews.xml", source = "xml", schema = customSchema, rowTag = "review")
saveDF(df, "/Volumes/<catalog>/<schema>/<volume>/newreviews.xml", "xml", "overwrite")

Чтение и запись XML с помощью SQL

Используйте DDL SQL для создания таблицы из XML-файла. Azure Databricks автоматически выводит типы столбцов.

DROP TABLE IF EXISTS reviews;
CREATE TABLE reviews
USING XML
OPTIONS (path "/Volumes/<catalog>/<schema>/<volume>/reviews.xml", rowTag "review");
SELECT * FROM reviews;

Можно также указать имена и типы столбцов в DDL. В этом случае схема не выводится автоматически.

DROP TABLE IF EXISTS reviews;

CREATE TABLE reviews (_id string, author string, rating integer, comment string)
USING XML
OPTIONS (path "/Volumes/<catalog>/<schema>/<volume>/reviews.xml", rowTag "review");

Загрузка XML с помощью COPY INTO

Используется COPY INTO для загрузки XML-файлов из облачного хранилища в таблицу Delta.

DROP TABLE IF EXISTS reviews;
CREATE TABLE IF NOT EXISTS reviews;

COPY INTO reviews
FROM "/Volumes/<catalog>/<schema>/<volume>/reviews.xml"
FILEFORMAT = XML
FORMAT_OPTIONS ('mergeSchema' = 'true', 'rowTag' = 'review')
COPY_OPTIONS ('mergeSchema' = 'true');

Чтение XML с проверкой строк

rowValidationXSDPath Используйте параметр для проверки каждой строки на основе схемы XSD во время чтения.

Питон

df = (spark.read
    .format("xml")
    .option("rowTag", "review")
    .option("rowValidationXSDPath", xsdPath)
    .load("/Volumes/<catalog>/<schema>/<volume>/reviews.xml"))
df.printSchema()

язык программирования Scala

val df = spark.read
  .option("rowTag", "review")
  .option("rowValidationXSDPath", xsdPath)
  .xml("/Volumes/<catalog>/<schema>/<volume>/reviews.xml")
df.printSchema

SQL

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews.xml',
  format => 'xml',
  rowTag => 'review',
  rowValidationXSDPath => '/Volumes/<catalog>/<schema>/<volume>/reviews.xsd'
)

Загрузка XML с помощью автозагрузчика

Используйте Auto Loader для непрерывной загрузки XML-файлов из облачного хранилища в таблицу Delta с автоматическим определением схемы и ее эволюцией.

Питон

query = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "xml")
  .option("rowTag", "review")
  .option("cloudFiles.inferColumnTypes", True)
  .option("cloudFiles.schemaLocation", schemaPath)
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .load(inputPath)
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", checkPointPath)
  .trigger(availableNow=True)
  .toTable("reviews")
)

язык программирования Scala

val query = spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "xml")
  .option("rowTag", "review")
  .option("cloudFiles.inferColumnTypes", true)
  .option("cloudFiles.schemaLocation", schemaPath)
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .load(inputPath)
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", checkPointPath)
  .trigger(Trigger.AvailableNow())
  .toTable("reviews")

Дополнительные ресурсы

  • Last updated on