Čtení a zápis souborů XML

Důležité

Tato funkce je ve verzi Public Preview.

Jazyk XML (Extensible Markup Language) je jazyk značek pro formátování, ukládání a sdílení dat v textovém formátu. Definuje sadu pravidel pro serializaci dat od dokumentů po libovolné datové struktury.

Azure Databricks podporuje XML pro čtení i zápis pomocí Apache Sparku, včetně automatického odvozování schématu a vývoje, konfigurace značek řádků, ověřování XSD a výrazů SQL, jako je from_xml. Nativní podpora XML funguje s Auto Loaderem, read_files a COPY INTO, bez nutnosti externích souborů JAR.

Předpoklady

Podpora formátu souborů XML vyžaduje Databricks Runtime 14.3 a vyšší.

Možnosti

Použijte metody .option() a .options() z DataFrameReader a DataFrameWriter ke konfiguraci zdrojů dat XML. Úplný seznam podporovaných možností najdete v tématu DataFrameReader Možnosti XML a DataFrameWriter možnosti XML.

Analýza záznamů XML

Specifikace XML vyžaduje dobře formátovanou strukturu. Tato specifikace se ale nehodí přímo do tabulkového formátu. Je nutné zadat volbu rowTag, která označuje prvek XML, který se mapuje na DataFrameRow. Prvek rowTag se stane nejvyšší úrovní struct. Podřízené prvky rowTag se stanou poli nejvyšší úrovně struct.

Můžete zadat schéma pro tento záznam nebo ho nechat automaticky odvodit. Vzhledem k tomu, že analyzátor zkoumá rowTag pouze prvky, odfiltrují se DTD a externí entity.

Následující příklady ilustrují odvozování schématu a parsování souboru XML pomocí různých rowTag možností:

Python

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)

Načtěte soubor XML s možností rowTag jako "reviews":

Python

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'
)

Výstup:

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}] |
+----------------------------------------------------------------------------------------+

Načtěte soubor XML pomocí rowTag jako "review":

Python

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'
)

Výstup:

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     |
+----+------+--------------------------------+------+

Ověření záznamů XML pomocí XSD

Volitelně můžete ověřit každý záznam XML na úrovni řádků definicí schématu XML (XSD). Soubor XSD je zadán v možnosti rowValidationXSDPath. XSD jinak neovlivňuje poskytnuté nebo odvozené schéma. Záznam, který selže, je označený jako poškozený a zpracován na základě možnosti režimu zpracování poškozených záznamů popsaných v části možnosti.

Můžete použít XSDToSchema k extrakci schématu datového rámce Spark ze souboru XSD. Podporuje pouze jednoduché, složité a sekvenční typy a podporuje pouze základní funkce 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))

Následující tabulka ukazuje převod datových typů XSD na datové typy Spark:

Datové typy XSD	Datové typy Sparku
`boolean`	`BooleanType`
`decimal`	`DecimalType`
`unsignedLong`	`DecimalType(38, 0)`
`double`	`DoubleType`
`float`	`FloatType`
`byte`	`ByteType`
`short`, `unsignedByte`	`ShortType`
`integer`, `negativeInteger`, `nonNegativeInteger`, `nonPositiveInteger`, , `positiveIntegerunsignedShort`	`IntegerType`
`long`, `unsignedInt`	`LongType`
`date`	`DateType`
`dateTime`	`TimestampType`
`Others`	`StringType`

Parsování vnořeného XML

Data XML ve sloupci s řetězcovými hodnotami v existujícím objektu DataFrame lze parsovat pomocí schema_of_xml a from_xml, které vracejí schéma a naparsované výsledky jako nové sloupce struct. Data XML předaná jako argument schema_of_xml a from_xml musí se jednat o jeden správně formátovaný záznam XML.

schema_of_xml

Slouží schema_of_xml k odvození schématu Sparku z řetězce XML. Předejte výsledek do from_xml, aby bylo možné parsovat sloupce XML.

Syntaxe: schema_of_xml(xmlStr [, options])

Argument	Required	Description
`xmlStr`	Yes	Výraz STRING určující jeden správně formátovaný záznam XML.
`options`	No	Literál `MAP<STRING,STRING>`, určující direktivy.

Vrátí hodnotu STRING, která obsahuje definici struktury s n poli řetězců, kde jsou názvy sloupců odvozeny od elementu XML a názvů atributů. Hodnoty polí obsahují odvozené formátované typy SQL.

from_xml

Pomocí from_xml lze parsovat sloupec typu STRING obsahující záznamy XML na typ struct. Zadejte schéma přímo nebo použijte výstup z schema_of_xml.

Syntaxe: from_xml(xmlStr, schema [, options])

Argument	Required	Description
`xmlStr`	Yes	Výraz STRING určující jeden správně formátovaný záznam XML.
`schema`	Yes	Výraz STRING nebo volání funkce `schema_of_xml`.
`options`	No	Literál `MAP<STRING,STRING>` určující direktivy.

Vrátí strukturu s názvy polí a typy odpovídající definici schématu. Schéma musí být definováno jako názvy sloupců oddělených čárkami a páry datových typů, které se používají například CREATE TABLE. Většina možností zobrazených v části Možnosti se dá použít s následujícími výjimkami:

rowTag: Protože existuje pouze jeden záznam XML, rowTag možnost není použitelná.
mode (výchozí: PERMISSIVE): Umožňuje režim zpracování poškozených záznamů během analýzy.
- PERMISSIVE: Když narazí na poškozený záznam, umístí chybný řetězec do pole nakonfigurovaného pomocí columnNameOfCorruptRecord a chybně formátovaná pole nastaví na null. Chcete-li zachovat poškozené záznamy, můžete nastavit pole typu řetězce pojmenované columnNameOfCorruptRecord v uživatelsky definovaném schématu. Pokud schéma pole neobsahuje, během analýzy zahodí poškozené záznamy. Při odvození schématu implicitně přidá do výstupního schématu pole columnNameOfCorruptRecord.
- FAILFAST: Vyvolá výjimku, když narazí na poškozené záznamy.

Příklady

Pokud chcete parsovat sloupec řetězce XML, použijte schema_of_xml k odvození schématu a pak ho předejte:from_xml

Python

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()

Parsování vloženého XML v 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>')
);

Převod mezi strukturami XML a datových rámců

Vzhledem k rozdílům ve struktuře mezi datovým rámcem DataFrame a XML existují určitá pravidla převodu z dat XML do DataFrame a z DataFrame dat XML. Všimněte si, že zpracování atributů lze zakázat pomocí možnosti excludeAttribute.

Převod z XML na datový rámec

Při čtení XML mapuje Azure Databricks elementy a atributy XML na pole datového rámce podle následujících pravidel.

Atributy jsou převedeny jako pole s předponou nadpisu attributePrefix.

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

Výsledkem je následující schéma:

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

Znaková data v elementu obsahujícím atributy nebo podřízené elementy se parsují do pole valueTag. Pokud existuje více výskytů znakových dat, valueTag pole se převede na array typ.

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

Výsledkem je následující schéma:

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)

Převod z datového rámce na XML

Při zápisu datového rámce do XML vyžadují určité vnořené struktury speciální zpracování z důvodu rozdílů mezi datovými modely DataFrame a XML.

Pokud DataFrame obsahuje pole ArrayType, jehož typ prvku je také ArrayType, při zápisu do XML se vytvoří další úroveň vnoření, která se při opětovném načtení a uložení souborů XML nevyskytuje. To má vliv jenom na datové rámce zdrojové mimo XML – čtení a zápis souborů XML zachovává původní strukturu.

Například datový rámec s následujícím schématem:

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

a následující údaje:

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

vytvoří následující výstup XML:

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

Název prvku nepojmenovaného pole v poli DataFrame je určen možností arrayElementName (Výchozí: item).

Povolte sloupec se zachráněnými daty

Sloupec zachráněných dat zajišťuje, že během ETL nikdy nepřijdete o data. Zaznamenává všechna data, která nebyla analyzována, protože jedno nebo více polí v záznamu má jeden z následujících problémů:

Chybí ze zadaného schématu.
Neodpovídá datovému typu zadaného schématu.
Neshoduje se s názvy polí v zadaném schématu.

Uložený datový sloupec se vrátí jako dokument JSON obsahující sloupce, které byly uloženy, a cestu ke zdrojovému souboru záznamu.

Chcete-li povolit sloupec pro zachráněná data, nastavte při čtení možnost rescuedDataColumn na název sloupce:

Python

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'
)

Pokud chcete odebrat cestu ke zdrojovému souboru ze sloupce zachráněných dat, nastavte:

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

Analyzátor XML podporuje při analýze záznamů tři režimy: PERMISSIVE, DROPMALFORMEDa FAILFAST. Při použití společně s datovým rescuedDataColumntypem neshody nezpůsobí vyřazení záznamů v DROPMALFORMED režimu nebo vyvolání chyby v FAILFAST režimu. Pouze poškozené záznamy (neúplné nebo chybně formátované XML) jsou zahazovány nebo způsobují chyby.

Odvozování a vývoj schématu pomocí Auto Loader

Podrobnou diskuzi o tomto tématu a příslušných možnostech najdete v tématu Konfigurace odvozování schématu a vývoje v auto loaderu. Automatický zavaděč můžete nakonfigurovat tak, aby automaticky rozpoznal schéma načtených dat XML, což umožňuje inicializovat tabulky bez explicitního deklarování schématu dat a vyvíjet schéma tabulky při zavádění nových sloupců. To eliminuje potřebu ručního sledování a použití změn schématu v průběhu času.

Ve výchozím nastavení se při odvozování schématu automatického zavaděče snaží vyhnout problémům s vývojem schématu kvůli neshodám typů. U formátů, které nekódují datové typy (JSON, CSV a XML), auto loader odvodí všechny sloupce jako řetězce, včetně vnořených polí v souborech XML. Apache Spark DataFrameReader používá jiné chování pro odvozování schématu a výběr datových typů pro sloupce ve zdrojích XML na základě ukázkových dat. Chcete-li toto chování povolit pomocí Auto Loaderu, nastavte možnost cloudFiles.inferColumnTypes na true.

Auto Loader zjistí přidání nových sloupců při zpracování dat. Když Auto Loader zjistí nový sloupec, datový proud se zastaví s UnknownFieldException. Než váš datový proud vyvolá tuto chybu, Auto Loader provede odvozování schématu na nejnovější mikrodávce dat a aktualizuje lokaci schématu pomocí nejnovějšího schématu sloučením nových sloupců na konec schématu. Datové typy existujících sloupců zůstávají beze změny. Auto Loader podporuje různé režimy pro vývoj schématu, který jste nastavili v možnosti cloudFiles.schemaEvolutionMode.

Pomocí nápověd pro schéma můžete u odvozeného schématu prosadit informace o schématu, které znáte a očekáváte. Pokud víte, že sloupec je konkrétní datový typ, nebo pokud chcete zvolit obecnější datový typ (například dvojité místo celého čísla), můžete zadat libovolný počet tipů pro datové typy sloupců jako řetězec pomocí syntaxe specifikace schématu SQL. Pokud je povolen sloupec zachráněných dat, pole pojmenovaná s odlišnou velikostí písmen, než je uvedeno ve schématu, se načtou do sloupce _rescued_data. Toto chování můžete změnit nastavením možnosti readerCaseSensitive na false, v takovém případě Auto Loader čte data bez rozlišování velkých a malých písmen.

Usage

Následující příklady používají datovou sadu Wanderbricks k předvedení čtení a zápisu souborů XML pomocí rozhraní API datového rámce Sparku a SQL.

Čtení a zápis XML

Pomocí rozhraní API datového rámce můžete psát recenze Wanderbricks do XML a číst je zpět.

Python

# 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()

R

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

Při čtení dat můžete schéma zadat ručně:

Python

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()

R

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")

Čtení a zápis XML pomocí SQL

K vytvoření tabulky ze souboru XML použijte JAZYK SQL DDL. Azure Databricks odvodí typy sloupců automaticky.

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

V DDL můžete také zadat názvy a typy sloupců. V tomto případě není schéma odvozeno automaticky.

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");

Načtení XML pomocí COPY INTO

Slouží COPY INTO k načtení souborů XML z cloudového úložiště do tabulky 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');

Čtení XML s ověřením řádků

rowValidationXSDPath Pomocí možnosti můžete při čtení ověřit každý řádek proti schématu XSD.

Python

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'
)

Načíst XML pomocí Auto Loader

Pomocí Auto Loaderu můžete průběžně načítat soubory XML z cloudového úložiště do tabulky Delta s automatickým odvozováním schématu a jeho evolucí.

Python

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")

Další materiály

Čtení a zápis dat XML pomocí spark-xml knihovny: Informace o uživatelích, kteří dříve používali opensourcovou knihovnu Spark XML, najdete v průvodci starší verzí integrace.
Čtení a zápis souborů JSON: Pokud jsou vaše data částečně strukturovaná, ale ne XML, JSON poskytuje podobnou podporu odvozování schématu a vnořených dat s jednodušším formátem.

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-06-24