Aller au contenu principal
Dernière mise à jour le

Lire et écrire des fichiers XML

info

Aperçu

Cette fonctionnalité est en aperçu public.

Extensible Markup Language (XML) est un langage de balisage pour le formatage, le stockage et le partage des données au format textuel. Il définit un ensemble de règles pour la sérialisation des données, allant des documents aux structures de données arbitraires.

Databricks prend en charge XML pour la lecture et l'écriture avec Apache Spark, y compris l'inférence et l'évolution automatiques de schéma, la configuration des balises de ligne, la validation XSD et les expressions SQL comme from_xml. La prise en charge native d'XML fonctionne avec Auto Loader, read_files et COPY INTO sans nécessiter de JAR externes.

Prérequis​

La prise en charge du format de fichier XML nécessite Databricks Runtime 14.3 et versions ultérieures.

Options​

Utilisez les méthodes .option() et .options() de DataFrameReader et DataFrameWriter pour configurer les sources de données XML. Pour une liste complète des options prises en charge, consultez DataFrameReader options XML et DataFrameWriter options XML.

Analysez les enregistrements XML​

La spécification XML exige une structure bien formée. Cependant, cette spécification ne correspond pas immédiatement à un format tabulaire. Vous devez spécifier l'option rowTag pour indiquer l'élément XML qui correspond à un DataFrame Row. L'élément rowTag devient le struct de niveau supérieur. Les éléments enfants de rowTag deviennent les champs du struct de niveau supérieur.

Vous pouvez spécifier le schéma pour cet enregistrement ou le laisser être inféré automatiquement. Étant donné que l'analyseur examine uniquement les éléments rowTag, les DTD et les entités externes sont filtrées.

Les exemples suivants illustrent l'inférence de schéma et l'analyse d'un fichier XML à l'aide de différentes options rowTag :

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)

Lisez le fichier XML avec l'option rowTag comme "reviews":

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

Résultat :

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

Lisez le fichier XML avec rowTag comme "review":

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

Résultat :

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

Valider les enregistrements XML avec XSD​

Vous pouvez éventuellement valider chaque enregistrement XML de niveau ligne par une définition de schéma XML (XSD). Le fichier XSD est spécifié dans l'option rowValidationXSDPath. Le XSD n'affecte pas le schéma fourni ou inféré. Un enregistrement qui échoue à la validation est marqué comme « corrompu » et géré en fonction de l'option de mode de gestion des enregistrements corrompus décrite dans la section des options.

Vous pouvez utiliser XSDToSchema pour extraire un schéma de DataFrame Spark d'un fichier XSD. Il prend en charge uniquement les types simples, complexes et de séquence, et prend en charge uniquement les fonctionnalités XSD de base.

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

Le tableau suivant montre la conversion des types de données XSD en types de données Spark :

Types de données XSD

Types de données 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

Types de données XSD

Types de données 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

Analysez un XML imbriqué​

Les données XML d'une colonne de type chaîne de caractères dans un DataFrame existant peuvent être analysées avec schema_of_xml et from_xml qui renvoient le schéma et les résultats analysés sous forme de nouvelles colonnes struct. Les données XML passées en argument à schema_of_xml et from_xml doivent être un enregistrement XML unique et bien formé.

schéma XML​

Utilisez schema_of_xml pour déduire le schéma Spark à partir d'une chaîne XML. Passez le résultat à from_xml pour analyser les colonnes XML.

Syntaxe : schema_of_xml(xmlStr [, options])

Argument

Obligatoire

Description

xmlStr

Oui

Une expression STRING spécifiant un seul enregistrement XML bien formé.

options

Non

Un littéral MAP<STRING,STRING> spécifiant des directives.

Argument

Obligatoire

Description

xmlStr

Oui

Une expression STRING spécifiant un seul enregistrement XML bien formé.

options

Non

Un littéral MAP<STRING,STRING> spécifiant des directives.

Renvoie une STRING contenant la définition d'une structure avec n champs de chaînes où les noms de colonnes sont dérivés de l'élément XML et des noms d'attributs. Les valeurs de champ contiennent les types SQL formatés dérivés.

from_xml​

Utilisez from_xml pour analyser une colonne de type chaîne de caractères contenant des enregistrements XML dans une structure. Fournissez un schéma directement ou utilisez la sortie de schema_of_xml.

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

Argument

Obligatoire

Description

xmlStr

Oui

Une expression STRING spécifiant un seul enregistrement XML bien formé.

schema

Oui

Une expression STRING ou invocation de la fonction schema_of_xml.

options

Non

Un littéral MAP<STRING,STRING> spécifiant des directives.

Argument

Obligatoire

Description

xmlStr

Oui

Une expression STRING spécifiant un seul enregistrement XML bien formé.

schema

Oui

Une expression STRING ou invocation de la fonction schema_of_xml.

options

Non

Un littéral MAP<STRING,STRING> spécifiant des directives.

Renvoie une structure avec des noms de champs et des types correspondant à la définition du schéma. Le schéma doit être défini comme des paires de noms de colonne et de types de données séparées par des virgules, comme utilisé dans, par exemple, CREATE TABLE. La plupart des options affichées dans la section Options sont applicables, sous réserve des exceptions suivantes :

  • rowTagPuisqu'il n'y a qu'un seul enregistrement XML, l'option rowTag n'est pas applicable.
  • mode (default : PERMISSIVE) : Permet un mode de gestion des enregistrements corrompus lors de l'analyse.
    • PERMISSIVE: Lorsqu'il rencontre un enregistrement corrompu, place la chaîne mal formée dans un champ configuré par columnNameOfCorruptRecord et définit les champs mal formés sur null. Pour conserver les enregistrements corrompus, vous pouvez définir un champ de type chaîne nommé columnNameOfCorruptRecord dans un schéma défini par l'utilisateur. Si un schéma ne contient pas le champ, il supprime les enregistrements corrompus lors de l'analyse. Lors de l'inférence d'un schéma, il ajoute implicitement un champ columnNameOfCorruptRecord dans un schéma de sortie.
    • FAILFAST: Lève une exception lorsqu'elle rencontre des enregistrements corrompus.

Exemples​

Pour analyser une colonne de chaîne XML, utilisez schema_of_xml pour déduire le schéma, puis passez-le à 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()

Pour analyser le XML en ligne en SQL :

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

Convertir entre les structures XML et DataFrame​

En raison des différences de structure entre un DataFrame et XML, il existe des règles de conversion des données XML vers DataFrame et de DataFrame vers les données XML. Notez que la gestion des attributs peut être désactivée avec l’option excludeAttribute.

Conversion de XML en DataFrame​

Lors de la lecture de XML, Databricks mappe les éléments et attributs XML aux champs de DataFrame selon les règles suivantes.

Les attributs sont convertis en champs avec le préfixe d'en-tête attributePrefix.

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

Ceci produit le schéma suivant :

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

Les données de caractères dans un élément contenant des attributs ou des éléments enfants sont analysées dans le champ valueTag. S'il existe plusieurs occurrences de données de caractères, le champ valueTag est converti en type array.

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

Ceci produit le schéma suivant :

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)

Conversion de DataFrame en XML​

Lors de l'écriture d'un DataFrame au format XML, certaines structures imbriquées nécessitent un traitement spécial en raison des différences entre les modèles de données DataFrame et XML.

Si un DataFrame contient un champ ArrayType dont le type d'élément est également ArrayType, l'écriture dans XML produit un niveau d'imbrication supplémentaire qui n'est pas présent lors de l'aller-retour des fichiers XML. Cela n'affecte que les DataFrames provenant de l'extérieur du XML — la lecture et l'écriture de fichiers XML préservent la structure originale.

Par exemple, un DataFrame avec le schéma suivant :

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

et les données suivantes :

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

produit la sortie XML suivante :

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

Le nom de l'élément du tableau sans nom dans le DataFrame est spécifié par l'option arrayElementName (default : item).

Activer la colonne de données récupérées​

La colonne de données récupérées garantit que vous ne perdez jamais de données pendant l'ETL. Il capture toutes les données qui n'ont pas été analysées car un ou plusieurs champs d'un enregistrement présentent l'un des problèmes suivants :

  • Absent du schéma fourni.
  • Ne correspond pas au type de données du schéma fourni.
  • Il y a une incohérence de casse avec les noms des champs dans le schéma fourni.

La colonne des données sauvées est renvoyée sous forme de document JSON contenant les colonnes qui ont été sauvées, et le chemin de fichier source de l'enregistrement.

Pour activer la colonne de données récupérées, définissez l'option rescuedDataColumn sur un nom de colonne lors de la lecture :

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

Pour supprimer le chemin du fichier source de la colonne de données récupérées, définissez :

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

L'analyseur XML prend en charge trois modes lors de l'analyse des enregistrements : PERMISSIVE, DROPMALFORMED et FAILFAST. Lorsqu'elles sont utilisées avec rescuedDataColumn, les non-correspondances de type de données n'entraînent pas la suppression d'enregistrements en mode DROPMALFORMED ou le déclenchement d'une erreur en mode FAILFAST. Seuls les enregistrements corrompus (XML incomplet ou malformé) sont supprimés ou génèrent des erreurs.

Déduire et faire évoluer le schéma avec Auto Loader​

Pour une discussion détaillée de ce sujet et des options applicables, consultez configurer l'inférence et l'évolution du schéma dans Auto Loader. Vous pouvez configurer Auto Loader pour détecter automatiquement le schéma des données XML chargées, vous permettant d'initialiser des tables sans déclarer explicitement le schéma de données et de faire évoluer le schéma de la table à mesure que de nouvelles colonnes sont introduites. Ceci élimine le besoin de suivre et d'appliquer manuellement les modifications de schéma au fil du temps.

default, l'inférence de schémas d'Auto Loader vise à éviter les problèmes d'évolution des schémas dus à des incompatibilités de types. Pour les formats qui n'encodent pas les types de données (JSON, CSV et XML), Auto Loader infère toutes les colonnes comme des chaînes, y compris les champs imbriqués dans les fichiers XML. L'Apache Spark DataFrameReader utilise un comportement différent pour l'inférence de schéma, sélectionnant les types de données pour les colonnes dans les sources XML basées sur des données d'échantillon. Pour activer ce comportement avec Auto Loader, définissez l'option cloudFiles.inferColumnTypes sur true.

Auto Loader détecte l'ajout de nouvelles colonnes lorsqu'il traite vos données. Quand Auto Loader détecte une nouvelle colonne, le Stream s'arrête avec un UnknownFieldException. Avant que votre Stream ne génère cette erreur, Auto Loader effectue l'inférence de schéma sur le dernier micro-batch de données et met à jour l'emplacement du schéma avec le dernier schéma en fusionnant les nouvelles colonnes à la fin du schéma. Les types de données des colonnes existantes restent inchangés. Auto Loader prend en charge différents modes d'évolution des schémas, que vous définissez dans l'option cloudFiles.schemaEvolutionMode.

Vous pouvez utiliser les indices de schéma pour appliquer les informations de schéma que vous connaissez et attendez sur un schéma inféré. Lorsque vous savez qu'une colonne est d'un type de données spécifique, ou si vous souhaitez choisir un type de données plus général (par exemple, un double au lieu d'un entier), vous pouvez fournir un nombre arbitraire d'indices pour les types de données de colonne sous forme de chaîne en utilisant la syntaxe de spécification de schéma SQL. Lorsque la colonne des données récupérées est activée, les champs nommés avec une casse différente de celle du schéma sont chargés dans la colonne _rescued_data. Vous pouvez modifier ce comportement en définissant l'option readerCaseSensitive sur false, auquel cas Auto Loader lit les données de manière insensible à la casse.

Utilisation​

Les exemples suivants utilisent le dataset Wanderbricks pour démontrer la lecture et l'écriture de fichiers XML à l'aide de l'API Spark DataFrame et de SQL.

Lecture et écriture XML​

Utilisez l'API DataFrame pour écrire et relire les avis Wanderbricks au format XML.

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

Vous pouvez spécifier manuellement le schéma lors de la lecture des données :

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

Lire et écrire du XML avec SQL​

Utilisez le DDL SQL pour créer une table à partir d'un fichier XML. Databricks infère automatiquement les types de colonnes.

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

Vous pouvez également spécifier les noms et types de colonnes en DDL. Dans ce cas, le schéma n'est pas inféré automatiquement.

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

Charger du XML à l’aide de COPY INTO​

Utilisez COPY INTO pour charger des fichiers XML depuis le stockage dans le cloud dans une table Delta.

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

Lire le XML avec validation de ligne​

Utilisez l’option rowValidationXSDPath pour valider chaque ligne par rapport à un schéma XSD pendant la lecture.

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

Charger du XML avec Auto Loader​

Utilisez Auto Loader pour ingérer en continu des fichiers XML depuis le stockage cloud dans une table Delta avec inférence et évolution automatiques du schéma.

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

Ressources supplémentaires​

Sur cette page