Notatka
Dostęp do tej strony wymaga autoryzacji. Może spróbować zalogować się lub zmienić katalogi.
Dostęp do tej strony wymaga autoryzacji. Możesz spróbować zmienić katalogi.
Ważne
Ta funkcja jest dostępna w publicznej wersji zapoznawczej.
Extensible Markup Language (XML) to język znaczników do formatowania, przechowywania i udostępniania danych w formacie tekstowym. Definiuje zestaw reguł serializacji danych, od dokumentów do dowolnych struktur danych.
Azure Databricks obsługuje kod XML do odczytywania i zapisywania przy użyciu platformy Apache Spark, w tym automatycznego wnioskowania schematu i ewolucji, konfiguracji tagów wierszy, walidacji XSD i wyrażeń SQL, takich jak from_xml. Natywna obsługa XML współpracuje z Auto Loader, read_files i COPY INTO bez konieczności używania zewnętrznych plików JAR.
Wymagania wstępne
Obsługa formatu pliku XML wymaga środowiska Databricks Runtime 14.3 lub nowszego.
Opcje
.option() Użyj metod .options() i DataFrameReader i DataFrameWriter , aby skonfigurować źródła danych XML. Aby uzyskać pełną listę obsługiwanych opcji, zobacz DataFrameReader Opcje XML i DataFrameWriter opcje XML.
Analizowanie rekordów XML
Specyfikacja XML nakazuje dobrze sformułowaną strukturę. Jednak ta specyfikacja nie przekłada się bezpośrednio na format tabelaryczny. Należy określić opcję rowTag, aby wskazać element XML, który odpowiada elementowi DataFrameRow. Element rowTag staje się najwyższym poziomem struct. Elementy podrzędne rowTag stają się polami najwyższego poziomu struct.
Możesz określić schemat dla tego rekordu lub zezwolić na automatyczne wnioskowanie. Ponieważ parser sprawdza tylko elementy rowTag, deklaracje DTD i encje zewnętrzne są odfiltrowywane.
W poniższych przykładach przedstawiono wnioskowanie schematu i analizowanie pliku XML przy użyciu różnych opcji 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)
Skala
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)
Odczytaj plik XML za pomocą opcji rowTag jako "reviews":
Python
df = spark.read.option("rowTag", "reviews").format("xml").load(xmlPath)
df.printSchema()
df.show(truncate=False)
Skala
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'
)
Wyjście:
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}] |
+----------------------------------------------------------------------------------------+
Odczytaj plik XML za 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:
Skala
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'
)
Wyjście:
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 |
+----+------+--------------------------------+------+
Weryfikowanie rekordów XML przy użyciu XSD
Opcjonalnie można zweryfikować każdy rekord XML na poziomie wiersza za pomocą definicji schematu XML (XSD). Plik XSD jest określony w rowValidationXSDPath opcji . XSD nie ma w inny sposób wpływu na podany lub wywnioskowany schemat. Rekord, który kończy się niepowodzeniem walidacji, jest oznaczony jako "uszkodzony" i obsługiwany na podstawie opcji trybu obsługi uszkodzonych rekordów opisanych w sekcji opcji.
Możesz użyć XSDToSchema do wyodrębnienia schematu ramki danych Spark z pliku XSD. Obsługuje tylko proste, złożone i sekwencyjne typy i obsługuje tylko podstawowe funkcje 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))
W poniższej tabeli przedstawiono konwersję typów danych XSD na typy danych platformy Spark:
| Typy danych XSD | Typy danych platformy Spark |
|---|---|
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 |
Analizowanie zagnieżdżonego kodu XML
Dane XML w kolumnie typu ciąg wartościowy w istniejącej tabeli danych mogą być analizowane przy użyciu schema_of_xml i from_xml, co zwraca schemat i przeanalizowane wyniki jako nowe kolumny struct. Dane XML przekazywane jako argument do schema_of_xml i from_xml muszą być pojedynczym dobrze sformułowanym rekordem XML.
schema_of_xml
Użyj schema_of_xml polecenia , aby wywnioskować schemat Spark z ciągu XML. Przekaż wynik do from_xml, aby analizować kolumny XML.
Składnia: schema_of_xml(xmlStr [, options])
| Argument | Required | Description |
|---|---|---|
xmlStr |
Yes | Wyrażenie STRING określające pojedynczy poprawnie sformułowany rekord XML. |
options |
No | Literał MAP<STRING,STRING> określający dyrektywy. |
Zwraca ciąg zawierający definicję struktury z n polami ciągów, w których nazwy kolumn pochodzą z nazwy elementu XML i nazw atrybutów. Wartości pól przechowują pochodne sformatowane typy SQL.
from_xml
Użyj from_xml, aby sparsować kolumnę typu STRING zawierającą rekordy XML do struktury. Podaj schemat bezpośrednio lub użyj wyniku działania schema_of_xml.
Składnia: from_xml(xmlStr, schema [, options])
| Argument | Required | Description |
|---|---|---|
xmlStr |
Yes | Wyrażenie STRING określające pojedynczy poprawnie sformułowany rekord XML. |
schema |
Yes | Wyrażenie STRING lub wywołanie funkcji schema_of_xml. |
options |
No | Literał MAP<STRING,STRING> określający dyrektywy. |
Zwraca strukturę z nazwami pól i typami pasującymi do definicji schematu. Schemat musi być zdefiniowany jako nazwa kolumny rozdzielanej przecinkami i pary typów danych, które są używane na przykład CREATE TABLE. Większość opcji pokazanych w sekcji Opcje ma zastosowanie z następującymi wyjątkami:
-
rowTag: Ponieważ istnieje tylko jeden rekord XML,rowTagopcja nie ma zastosowania. -
mode(ustawienie domyślne:PERMISSIVE): umożliwia tryb radzenia sobie z uszkodzonymi rekordami podczas analizowania.-
PERMISSIVE: Gdy napotka uszkodzony rekord, umieszcza nieprawidłowy ciąg w polu skonfigurowanym przezcolumnNameOfCorruptRecordi ustawia nieprawidłowe pola nanull. Aby zachować uszkodzone rekordy, można ustawić pole typu ciągu o nazwiecolumnNameOfCorruptRecordw schemacie zdefiniowanym przez użytkownika. Jeśli schemat nie ma pola, usuwa uszkodzone rekordy podczas analizowania. Podczas wnioskowania schematu niejawnie dodaje polecolumnNameOfCorruptRecordw schemacie wyjściowym. -
FAILFAST: zgłasza wyjątek, gdy napotyka uszkodzone rekordy.
-
Przykłady
Aby przeanalizować kolumnę zawierającą ciąg XML, użyj schema_of_xml, aby wywnioskować schemat, a następnie przekaż go do 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()
Skala
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()
Aby przeanalizować wbudowany kod XML w języku 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>')
);
Konwertowanie między strukturami XML i DataFrame
Ze względu na różnice w strukturze między DataFrame a XML istnieją pewne reguły konwersji danych XML do DataFrame oraz z DataFrame do danych XML. Należy pamiętać, że atrybuty obsługi można wyłączyć za pomocą opcji excludeAttribute.
Konwersja z xml na ramkę danych
Podczas odczytywania kodu XML Azure Databricks mapuje elementy XML i atrybuty na pola ramki danych zgodnie z następującymi regułami.
Atrybuty są konwertowane jako pola z prefiksem attributePrefixnagłówka .
<one myOneAttrib="AAAA">
<two>two</two>
<three>three</three>
</one>
Spowoduje to utworzenie następującego schematu:
root
|-- _myOneAttrib: string (nullable = true)
|-- two: string (nullable = true)
|-- three: string (nullable = true)
Dane znaków w elemecie zawierającym atrybuty lub elementy podrzędne są analizowane w valueTag polu. Jeśli występuje wiele danych znakowych, pole valueTag jest konwertowane do typu array.
<one>
<two myTwoAttrib="BBBBB">two</two>
some value between elements
<three>three</three>
some other value between elements
</one>
Spowoduje to utworzenie następującego schematu:
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)
Konwersja z ramki danych na XML
Podczas zapisywania ramki danych w formacie XML niektóre zagnieżdżone struktury wymagają specjalnej obsługi ze względu na różnice między modelami danych DataFrame i XML.
Jeśli obiekt DataFrame zawiera pole ArrayType, którego typem elementów jest również ArrayType, zapisanie go do formatu XML powoduje utworzenie dodatkowego poziomu zagnieżdżenia, którego nie ma podczas wielokrotnego odczytu i zapisu plików XML. Ma to wpływ tylko na ramki danych źródłowe poza kodem XML — odczytywanie i zapisywanie plików XML zachowuje oryginalną strukturę.
Na przykład ramka danych z następującym schematem:
|-- a: array (nullable = true)
| |-- element: array (containsNull = true)
| | |-- element: string (containsNull = true)
oraz następujące dane:
+------------------------------------+
| a|
+------------------------------------+
|[WrappedArray(aa), WrappedArray(bb)]|
+------------------------------------+
tworzy następujące dane wyjściowe XML:
<a>
<item>aa</item>
</a>
<a>
<item>bb</item>
</a>
Nazwa elementu tablicy bez nazwy w DataFrame jest określana przez opcję arrayElementName (Wartość domyślna: item).
Włącz kolumnę odzyskanych danych
Uratowana kolumna danych gwarantuje, że nigdy nie utracisz danych podczas etL. Przechwytuje wszystkie dane, które nie zostały przeanalizowane, ponieważ co najmniej jedno pole w rekordzie ma jeden z następujących problemów:
- Brak podanego schematu.
- Nie jest zgodny z typem danych podanego schematu.
- Ma niezgodność pisowni liter z nazwami pól w podanym schemacie.
Uratowana kolumna danych jest zwracana jako dokument JSON zawierający kolumny, które zostały uratowane, oraz ścieżkę pliku źródłowego rekordu.
Aby włączyć kolumnę odzyskanych danych, ustaw opcję rescuedDataColumn na nazwę kolumny podczas wczytywania:
Python
df = spark.read.option("rescuedDataColumn", "_rescued_data").format("xml").load("/Volumes/<catalog>/<schema>/<volume>/reviews_xml")
Skala
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'
)
Aby usunąć ścieżkę pliku źródłowego z uratowanej kolumny danych, ustaw:
spark.conf.set("spark.databricks.sql.rescuedDataColumn.filePath.enabled", "false")
Analizator XML obsługuje trzy tryby podczas analizowania rekordów: PERMISSIVE, DROPMALFORMEDi FAILFAST. Gdy jest używane razem z rescuedDataColumn, niezgodności typów danych nie powodują odrzucania rekordów w trybie DROPMALFORMED ani zgłoszenia błędu w trybie FAILFAST. Tylko uszkodzone rekordy (niekompletny lub niepoprawny składniowo XML) są odrzucane lub powodują błędy.
Wykrywanie i ewoluowanie schematu za pomocą Auto Loader
Aby szczegółowo omówić ten temat i odpowiednie opcje, zobacz Konfigurowanie wnioskowania schematu i ewolucji w automatycznym module ładującym. Możesz skonfigurować moduł automatycznego ładowania w celu automatycznego wykrywania schematu załadowanych danych XML, co umożliwia inicjowanie tabel bez jawnego deklarowania schematu danych i rozwijania schematu tabeli w miarę wprowadzania nowych kolumn. Eliminuje to konieczność ręcznego śledzenia i stosowania zmian schematu w czasie.
Domyślnie wnioskowanie schematu przez Auto Loader ma na celu unikanie problemów z ewolucją schematu wynikających z niezgodności typów. W przypadku formatów, które nie kodują typów danych (JSON, CSV i XML), moduł ładujący automatycznie wywnioskuje wszystkie kolumny jako ciągi, w tym pola zagnieżdżone w plikach XML. Usługa Apache Spark DataFrameReader używa innego zachowania do wnioskowania schematu, wybierając typy danych dla kolumn w źródłach XML na podstawie przykładowych danych. Aby włączyć to zachowanie za pomocą modułu automatycznego ładowania, ustaw opcję cloudFiles.inferColumnTypes na true.
Moduł automatycznego ładowania wykrywa dodanie nowych kolumn podczas przetwarzania danych. Gdy funkcja automatycznego ładowania wykryje nową kolumnę, strumień zatrzymuje się z kodem UnknownFieldException. Przed wystąpieniem tego błędu, Auto Loader przeprowadza wnioskowanie schematu na najnowszym mikropakiecie danych i aktualizuje lokalizację schematu przy użyciu najnowszego schematu, dodając nowe kolumny na końcu schematu. Typy danych istniejących kolumn pozostają niezmienione. Moduł automatycznego ładowania obsługuje różne tryby ewolucji schematu, które są ustawiane w opcji cloudFiles.schemaEvolutionMode.
Możesz użyć wskazówek schematu , aby wymusić informacje o schemacie, które znasz i których oczekujesz na wywnioskowanym schemacie. Jeśli wiesz, że kolumna ma określony typ danych lub jeśli chcesz wybrać bardziej ogólny typ danych (na przykład podwójne zamiast liczby całkowitej), możesz podać dowolną liczbę wskazówek dla typów danych kolumn jako ciąg przy użyciu składni specyfikacji schematu SQL. Po włączeniu uratowanej kolumny danych pola o nazwie w przypadku innym niż schemat są ładowane do kolumny _rescued_data. To zachowanie można zmienić, ustawiając opcję readerCaseSensitive na false; wówczas Auto Loader odczytuje dane bez rozróżniania wielkości liter.
Usage
W poniższych przykładach użyto zestawu danych usługi Wanderbricks do zademonstrowania odczytywania i zapisywania plików XML przy użyciu interfejsu API ramki danych platformy Spark i języka SQL.
Odczytywanie i zapisywanie kodu XML
Użyj interfejsu API DataFrame, aby zapisać recenzje Wanderbricks do formatu XML i odczytać je z powrotem.
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()
Skala
// 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")
Schemat można określić ręcznie podczas odczytywania danych:
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()
Skala
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")
Odczytywanie i zapisywanie kodu XML przy użyciu języka SQL
Użyj języka SQL DDL, aby utworzyć tabelę na podstawie pliku XML. Azure Databricks automatycznie wykrywa typy kolumn.
DROP TABLE IF EXISTS reviews;
CREATE TABLE reviews
USING XML
OPTIONS (path "/Volumes/<catalog>/<schema>/<volume>/reviews.xml", rowTag "review");
SELECT * FROM reviews;
Można również określić nazwy i typy kolumn w języku DDL. W takim przypadku schemat nie jest automatycznie wnioskowany.
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");
Ładowanie kodu XML przy użyciu COPY INTO
Użyj COPY INTO, aby załadować pliki XML z magazynu w chmurze do tabeli 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');
Odczytywanie kodu XML z walidacją wierszy
rowValidationXSDPath Użyj opcji , aby zweryfikować każdy wiersz względem schematu XSD podczas odczytywania.
Python
df = (spark.read
.format("xml")
.option("rowTag", "review")
.option("rowValidationXSDPath", xsdPath)
.load("/Volumes/<catalog>/<schema>/<volume>/reviews.xml"))
df.printSchema()
Skala
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'
)
Ładowanie kodu XML za pomocą modułu ładującego automatycznego
Użyj Auto Loader do ciągłego ładowania plików XML z magazynu w chmurze do tabeli Delta z automatycznym wykrywaniem i ewolucją schematu.
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")
)
Skala
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")
Dodatkowe zasoby
-
Odczytywanie i zapisywanie danych XML przy użyciu
spark-xmlbiblioteki: w przypadku użytkowników, którzy wcześniej korzystali z biblioteki XML platformy Spark typu open source, zobacz starszy przewodnik integracji. - Odczytywanie i zapisywanie plików JSON: jeśli dane są częściowo ustrukturyzowane, ale nie XML, kod JSON zapewnia podobne wnioskowanie schematu i obsługę danych zagnieżdżonych przy użyciu prostszego formatu.