Lesen und Schreiben von XML-Dateien

Wichtig

Dieses Feature befindet sich in der Public Preview.

Extensible Markup Language (XML) ist eine Markupsprache zum Formatieren, Speichern und Freigeben von Daten im Textformat. Sie definiert verschiedene Regeln zum Serialisieren von Daten, die von Dokumenten bis hin zu beliebigen Datenstrukturen reichen.

Azure Databricks unterstützt XML für das Lesen und Schreiben mit Apache Spark, einschließlich automatischer Schemainference und Evolution, Zeilentagkonfiguration, XSD-Validierung und SQL-Ausdrücke wie from_xml. Native XML-Unterstützung funktioniert mit auto Loader, read_filesund COPY INTO ohne externe Jars.

Voraussetzungen

Die XML-Dateiformatunterstützung erfordert Databricks Runtime 14.3 und höher.

Options

Verwenden Sie die Methoden .option() und .options() von DataFrameReader und DataFrameWriter, um XML-Datenquellen zu konfigurieren. Eine vollständige Liste der unterstützten Optionen finden Sie unter DataFrameReader XML-Optionen und DataFrameWriter XML-Optionen.

Parsen von XML-Datensätzen

Die XML-Spezifikation schreibt eine wohlgeformte Struktur vor. Diese Spezifikation wird jedoch nicht sofort einem tabellarischen Format zugeordnet. Sie müssen mit der Option rowTag das XML-Element angeben, das DataFrameRow zugeordnet wird. Das rowTag-Element wird zum struct der obersten Ebene. Die untergeordneten Elemente von rowTag werden zu den Feldern der Struktur (struct) der obersten Ebene.

Sie können das Schema für diesen Datensatz angeben oder es automatisch ableiten lassen. Da der Parser nur die rowTag-Elemente untersucht, werden DTD und externe Entitäten herausgefiltert.

Die folgenden Beispiele veranschaulichen den Schemarückschluss und die Analyse einer XML-Datei mit unterschiedlichen rowTag-Optionen:

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)

Lesen Sie die XML-Datei mit der Option rowTag als "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'
)

Ausgabe:

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

Lesen Sie die XML-Datei mit rowTag als "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'
)

Ausgabe:

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

Überprüfen von XML-Datensätzen mit XSD

Sie können optional jeden XML-Datensatz auf Zeilenebene anhand einer XML-Schemadefinition (XSD) überprüfen. Die XSD-Datei wird in der Option rowValidationXSDPath angegeben. Die XSD wirkt sich nicht anderweitig auf das bereitgestellte oder abgeleitete Schema aus. Ein Datensatz, der die Überprüfung fehlschlägt, wird als "beschädigt" gekennzeichnet und basierend auf der im Optionsabschnitt beschriebenen Option für den beschädigten Datensatzbehandlungsmodus behandelt.

Sie können XSDToSchema verwenden, um ein Spark DataFrame-Schema aus einer XSD-Datei zu extrahieren. Es unterstützt nur einfache, komplexe und sequentielle Typen und nur grundlegende XSD-Funktionen.

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

Die folgende Tabelle zeigt die Konvertierung von XSD-Datentypen in Spark-Datentypen:

XSD-Datentypen Spark-Datentypen
boolean BooleanType
decimal DecimalType
unsignedLong DecimalType(38, 0)
double DoubleType
float FloatType
byte ByteType
short, unsignedByte ShortType
integer, , negativeIntegernonNegativeInteger, nonPositiveInteger, , positiveIntegerunsignedShort IntegerType
long, unsignedInt LongType
date DateType
dateTime TimestampType
Others StringType

Parsen von verschachteltem XML

XML-Daten in einer Spalte mit Zeichenfolgenwerten in einem vorhandenen Datenrahmen können mit schema_of_xml und from_xml analysiert werden. Dabei werden das Schema und die analysierten Ergebnisse als neue struct-Spalten zurückgegeben. XML-Daten, die als Argument an schema_of_xml und from_xml übergeben werden, müssen ein einzelner wohlgeformter XML-Datensatz sein.

Schema von XML

Wird verwendet schema_of_xml , um das Spark-Schema aus einer XML-Zeichenfolge abzuleiten. Übergeben Sie das Ergebnis an from_xml, um XML-Spalten zu analysieren.

Syntax: schema_of_xml(xmlStr [, options])

Argument Erforderlich Description
xmlStr Yes Ein STRING-Ausdruck, der einen einzelnen wohlgeformten XML-Eintrag angibt.
options No Ein MAP<STRING,STRING>-Literal, das Direktiven angibt.

Gibt eine ZEICHENFOLGE zurück, die eine Definition einer Struktur mit n Zeichenfolgenfeldern enthält, in denen die Spaltennamen aus dem XML-Element und Attributnamen abgeleitet werden. Die Feldwerte enthalten die abgeleiteten formatierten SQL Typen.

from_xml

Verwenden Sie from_xml, um eine STRING-Spalte, die XML-Datensätze enthält, in eine Struct zu parsen. Geben Sie ein Schema direkt an, oder verwenden Sie die Ausgabe von schema_of_xml.

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

Argument Erforderlich Description
xmlStr Yes Ein STRING-Ausdruck, der einen einzelnen wohlgeformten XML-Eintrag angibt.
schema Yes Ein STRING-Ausdruck oder Aufruf der schema_of_xml-Funktion.
options No Ein MAP<STRING,STRING>-Literal, das Direktiven angibt.

Gibt eine Struktur mit Feldnamen und Typen zurück, die der Schemadefinition entsprechen. Das Schema muss als Spaltennamen- und Datentyppaare mit Kommas als Trennzeichen definiert werden, z. B wie in CREATE TABLE. Die meisten Optionen im Abschnitt "Optionen " gelten für die folgenden Ausnahmen:

  • rowTag: Da nur ein XML-Datensatz vorhanden ist, kann die Option rowTag nicht angewendet werden.
  • mode (Standardwert: PERMISSIVE): Ermöglicht einen Modus für den Umgang mit beschädigten Datensätzen während des Parsens.
    • PERMISSIVE: Wenn ein beschädigter Datensatz erkannt wird, wird die falsch formatierte Zeichenfolge in ein durch columnNameOfCorruptRecord konfiguriertes Feld eingefügt, und die falsch formatierten Felder werden auf null festgelegt. Um beschädigte Datensätze beizubehalten, können Sie ein Zeichenfolgenfeld namens columnNameOfCorruptRecord in einem benutzerdefinierten Schema festlegen. Wenn ein Schema das Feld nicht enthält, verwirft es beim Parsen beschädigte Datensätze. Beim Ableiten eines Schemas wird in einem Ausgabeschema implizit ein columnNameOfCorruptRecord-Feld hinzugefügt.
    • FAILFAST: Hiermit wird eine Ausnahme ausgelöst, wenn beschädigte Datensätze erkannt werden.

Beispiele

Um eine XML-Stringspalte zu parsen, verwenden Sie schema_of_xml, um das Schema abzuleiten, und übergeben Sie das Ergebnis dann an 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()

So analysieren Sie Inline-XML in 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>')
);

Konvertieren zwischen XML- und DataFrame-Strukturen

Aufgrund struktureller Unterschiede zwischen DataFrames und XML gibt es einige Konvertierungsregeln von XML-Daten in DataFrame und von DataFrame in XML-Daten. Beachten Sie, dass die Behandlung von Attributen mit der Option excludeAttribute deaktiviert werden kann.

Konvertieren von XML in DataFrame

Beim Lesen von XML ordnet Azure Databricks XML-Elemente und -Attribute den DataFrame-Feldern gemäß den folgenden Regeln zu.

Attribute werden als Felder mit dem Überschriftenpräfix attributePrefixkonvertiert.

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

Dadurch wird das folgende Schema erzeugt:

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

Zeichendaten in einem Element, das Attribute oder untergeordnete Elemente enthält, werden in das Feld valueTag geparst. Bei mehreren Vorkommen von Zeichendaten wird das Feld valueTag in einen array-Typ konvertiert.

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

Dadurch wird das folgende Schema erzeugt:

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)

Konvertieren von DataFrame in XML

Beim Schreiben eines DataFrame in XML erfordern bestimmte geschachtelte Strukturen aufgrund von Unterschieden zwischen DataFrame- und XML-Datenmodellen eine spezielle Behandlung.

Wenn ein DataFrame ein ArrayType-Feld enthält, dessen Elementtyp ebenfalls ArrayType ist, wird beim Schreiben als XML eine zusätzliche Schachtelungsebene erzeugt, die beim Roundtrip von XML-Dateien nicht vorhanden ist. Dies betrifft nur DataFrames, die außerhalb von XML stammen – das Lesen und Schreiben von XML-Dateien behält die ursprüngliche Struktur bei.

Beispiel: Ein DataFrame mit dem folgenden Schema:

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

und die folgenden Daten:

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

erzeugt die folgende XML-Ausgabe:

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

Der Elementname des nicht benannten Arrays in DataFrame wird durch die Option arrayElementName angegeben (Standardwert: item).

Aktivieren der Spalte "Gerettete Daten"

Die gerettete Datenspalte stellt sicher, dass Während der ETL niemals Daten verloren gehen. Es erfasst alle Daten, die nicht analysiert wurden, da eines oder mehrere Felder in einem Datensatz eines der folgenden Probleme aufweisen:

  • Fehlt im bereitgestellten Schema.
  • Stimmt nicht mit dem Datentyp des bereitgestellten Schemas überein.
  • Weist einen Fallkonflikt mit den Feldnamen im angegebenen Schema auf.

Die gerettete Datenspalte wird als JSON-Dokument zurückgegeben, das die geretteten Spalten und den Quelldateipfad des Datensatzes enthält.

Um die gerettete Datenspalte zu aktivieren, legen Sie beim Lesen die rescuedDataColumn Option auf einen Spaltennamen fest:

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

Wenn Sie den Quelldateipfad aus der Spalte für die geretteten Daten entfernen möchten, legen Sie Folgendes fest:

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

Der XML-Parser unterstützt drei Modi beim Parsen von Datensätzen: PERMISSIVE, DROPMALFORMED und FAILFAST. Bei Verwendung mit rescuedDataColumn führen Datentypkonflikte nicht dazu, dass Datensätze im Modus DROPMALFORMED gelöscht werden oder im Modus FAILFAST einen Fehler auslösen. Nur beschädigte Datensätze (unvollständige oder falsch formatierte XML-Elemente) werden verworfen oder führen zu Fehlern.

Schema mit Auto Loader ableiten und weiterentwickeln

Eine ausführliche Erörterung dieses Themas und der verfügbaren Optionen finden Sie unter Schemainferenz und -entwicklung in Auto Loader konfigurieren. Sie können den Autoloader so konfigurieren, dass das Schema der geladenen XML-Daten automatisch erkannt wird. Dadurch können Sie Tabellen initialisieren, ohne das Datenschema explizit zu deklarieren, und das Tabellenschema entwickeln, wenn neue Spalten eingeführt werden. Dadurch wird die Notwendigkeit beseitigt, Schemaänderungen im Laufe der Zeit manuell nachverfolgen und anwenden zu müssen.

Standardmäßig versucht die Schemainferenz von Auto Loader, Probleme bei der Schemaweiterentwicklung aufgrund von Typkonflikten zu vermeiden. Bei Formaten, die datentypen (JSON, CSV und XML) nicht codieren, leitet das automatische Laden alle Spalten als Zeichenfolgen ab, einschließlich geschachtelter Felder in XML-Dateien. Apache Spark DataFrameReader weist bei der Schemainferenz ein anderes Verhalten auf und wählt Datentypen für Spalten in XML-Quellen anhand von Beispieldaten aus. Um dieses Verhalten im Autoloader zu aktivieren, legen Sie die Option cloudFiles.inferColumnTypes auf true fest.

Autoloader erkennt das Hinzufügen neuer Spalten, während er Ihre Daten verarbeitet. Wenn Autoloader eine neue Spalte erkennt, wird der Stream mit einer UnknownFieldException beendet. Bevor Ihr Stream diesen Fehler auslöst, führt Auto Loader eine Schemainferenz für den neuesten Micro-Batch an Daten durch und aktualisiert den Speicherort des Schemas mit dem neuesten Schema, indem neue Spalten am Ende des Schemas angefügt werden. Die Datentypen vorhandener Spalten bleiben unverändert. Der Autoloader unterstützt die folgenden Modi für die Schemaentwicklung, die Sie in der Option cloudFiles.schemaEvolutionMode festlegen.

Sie können Schemahinweise verwenden, um die Schemainformationen, die Sie kennen und erwarten, für ein abgeleitetes Schema durchzusetzen. Wenn Sie wissen, dass eine Spalte einen bestimmten Datentyp hat, oder wenn Sie einen noch allgemeineren Datentyp wählen möchten (z. B. „double“ statt „integer“), können Sie eine beliebige Anzahl von Hinweisen für die Datentypen von Spalten als eine Zeichenfolge mithilfe der SQL-Schemaspezifikationssyntax angeben. Wenn die Spalte für wiederhergestellte Daten aktiviert ist, werden Felder, deren Name eine andere Groß-/Kleinschreibung als das Schema aufweist, in die Spalte _rescued_data geladen. Sie können dieses Verhalten ändern, indem Sie die Option readerCaseSensitive auf false festlegen. In diesem Fall werden vom Autoloader Daten gelesen, ohne dass zwischen Groß- und Kleinschreibung unterschieden wird.

Usage

In den folgenden Beispielen wird das Wanderbricks-Dataset verwendet, um das Lesen und Schreiben von XML-Dateien mithilfe der Spark DataFrame-API und SQL zu veranschaulichen.

Lesen und Schreiben von XML

Verwenden Sie die DataFrame-API, um Wanderbricks-Rezensionen in XML zu schreiben und sie zurückzulesen.

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

Sie können das Schema beim Lesen von Daten manuell angeben:

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

Lesen und Schreiben von XML mit SQL

Verwenden Sie SQL DDL, um eine Tabelle aus einer XML-Datei zu erstellen. Azure Databricks leitet Spaltentypen automatisch ab.

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

Sie können auch Spaltennamen und -typen in DDL angeben. In diesem Fall wird das Schema nicht automatisch abgeleitet.

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

Laden von XML mithilfe von COPY INTO

Wird COPY INTO verwendet, um XML-Dateien aus dem Cloudspeicher in eine Delta-Tabelle zu laden.

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

Lesen von XML mit Zeilenüberprüfung

Verwenden Sie die rowValidationXSDPath Option, um jede Zeile beim Lesen anhand eines XSD-Schemas zu überprüfen.

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

Laden von XML mit dem Autoloader

Verwenden Sie "Auto Loader", um XML-Dateien kontinuierlich aus dem Cloudspeicher in eine Delta-Tabelle mit automatischer Schemaference und Evolution aufzunehmen.

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

Zusätzliche Ressourcen

  • Last updated on