Ler e gravar arquivos XML

Importante

Esta funcionalidade está em Pré-visualização Pública.

Linguagem de Marcação Extensível (XML) é uma linguagem de marcação para formatar, armazenar e partilhar dados em formato textual. Ele define um conjunto de regras para serializar dados que variam de documentos a estruturas de dados arbitrárias.

O Azure Databricks suporta XML tanto para leitura como para escrita com o Apache Spark, incluindo inferência e evolução automática de esquemas, configuração de etiquetas de linha, validação XSD e expressões SQL como from_xml. O suporte nativo a XML funciona com o Auto Loader, read_files, e COPY INTO sem necessidade de jars externos.

Prerequisites

O suporte ao formato de ficheiro XML requer Databricks Runtime 14.3 e superiores.

Opções

Utilize os métodos .option() e .options() de DataFrameReader e DataFrameWriter para configurar origens de dados XML. Para uma lista completa de opções suportadas, consulte DataFrameReader opções XML e DataFrameWriter opções XML.

Analisar registros XML

A especificação XML exige uma estrutura bem formada. No entanto, essa especificação não é mapeada imediatamente para um formato tabular. Deve especificar a opção rowTag para indicar o elemento XML que corresponde a um ficheiro DataFrameRow. O rowTag elemento torna-se o nível structsuperior. Os elementos filho de rowTag tornam-se os campos do struct de nível superior.

Você pode especificar o esquema para esse registro ou deixá-lo ser inferido automaticamente. Como o analisador examina apenas os elementos rowTag, o DTD e as entidades externas são filtrados.

Os exemplos a seguir ilustram a inferência de esquema e a análise de um arquivo XML usando diferentes opções de 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)

linguagem de programação 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)

Leia o ficheiro XML com a opção rowTag como "reviews":

Python

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

linguagem de programação 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'
)

Saída:

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

Leia o ficheiro XML com rowTag como "review":

Python

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

linguagem de programação 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'
)

Saída:

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

Validar registos XML com XSD

Opcionalmente, você pode validar cada registro XML de nível de linha por uma definição de esquema XML (XSD). O arquivo XSD é especificado na rowValidationXSDPath opção. De outra forma, o XSD não afeta o esquema fornecido ou inferido. Um registo que falha na validação é marcado como "corrompido" e tratado com base na opção de modo de tratamento de registos corrompidos descrita na secção de opções.

Você pode usar XSDToSchema para extrair um esquema Spark DataFrame de um arquivo XSD. Ele suporta apenas tipos simples, complexos e de sequência, e suporta apenas a funcionalidade XSD básica.

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

A tabela a seguir mostra a conversão de tipos de dados XSD em tipos de dados do Spark:

Tipos de dados XSD Tipos de dados do 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

Analisar XML aninhado

Os dados XML numa coluna do tipo cadeia de caracteres num DataFrame existente podem ser processados por schema_of_xml e from_xml que retornam o esquema e os resultados analisados como novas colunas struct. Os dados XML passados como um argumento para schema_of_xml e from_xml devem ser um único registro XML bem formado.

esquema_de_xml

Use schema_of_xml para inferir o esquema Spark a partir de uma string XML. Passe o resultado para from_xml para analisar colunas XML.

Sintaxe: schema_of_xml(xmlStr [, options])

Argument Required Description
xmlStr Yes Uma expressão STRING que especifica um único registo XML bem formado.
options No Um MAP<STRING,STRING> literal que especifica diretivas.

Devolve uma STRING que contém uma definição de struct com n campos de cadeias de caracteres, em que os nomes das colunas derivam dos nomes dos elementos e dos atributos XML. Os valores de campo contêm os tipos SQL formatados derivados.

from_xml

Use from_xml para analisar uma coluna STRING contendo registos XML numa estrutura. Forneça um esquema diretamente ou use a saída de schema_of_xml.

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

Argument Required Description
xmlStr Yes Uma expressão STRING que especifica um único registo XML bem formado.
schema Yes Uma expressão STRING ou a invocação da função schema_of_xml.
options No Um MAP<STRING,STRING> literal que especifica diretivas.

Devolve uma estrutura com nomes de campos e tipos que correspondem à definição do esquema. O esquema deve ser definido como nome de coluna separado por vírgulas e pares de tipos de dados, conforme usado, por exemplo, em CREATE TABLE. A maioria das opções apresentadas na secção de Opções é aplicável, com as seguintes exceções:

  • rowTag: Como há apenas um registro XML, a rowTag opção não é aplicável.
  • mode (padrão: PERMISSIVE): Permite um modo para lidar com registros corrompidos durante a análise.
    • PERMISSIVE: Quando ele encontra um registro corrompido, coloca a cadeia de caracteres malformada em um campo configurado por columnNameOfCorruptRecord, e define campos malformados como null. Para manter registos corrompidos, pode-se definir um campo do tipo string chamado columnNameOfCorruptRecord num esquema definido pelo utilizador. Se um esquema não tiver o campo, ele descartará registros corrompidos durante a análise. Ao inferir um esquema, ele adiciona implicitamente um campo columnNameOfCorruptRecord em um esquema de saída.
    • FAILFAST: Lança uma exceção quando encontra registos corrompidos.

Exemplos

Para analisar uma coluna de strings XML, use schema_of_xml para inferir o esquema e depois passá-lo para 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()

linguagem de programação 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()

Para analisar XML inline em 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>')
);

Converter entre estruturas XML e DataFrame

Devido às diferenças de estrutura entre DataFrame e XML, existem algumas regras de conversão de dados XML para DataFrame e de DataFrame para dados XML. Observe que a manipulação de atributos pode ser desabilitada com a opção excludeAttribute.

Conversão de XML para DataFrame

Ao ler XML, o Azure Databricks mapeia elementos e atributos XML para campos DataFrame de acordo com as seguintes regras.

Os atributos são convertidos em campos com o prefixo de cabeçalho attributePrefix.

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

Isto produz o seguinte esquema:

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

Os dados de caracteres num elemento que contém atributo(s) ou elemento(s) filho(s) são processados e colocados no campo valueTag. Se houver várias ocorrências de dados de caracteres, o valueTag campo será convertido em um array tipo.

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

Isto produz o seguinte esquema:

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)

Conversão de DataFrame para XML

Ao escrever um DataFrame para XML, certas estruturas aninhadas requerem tratamento especial devido às diferenças entre os modelos de dados DataFrame e XML.

Se um DataFrame contiver um campo ArrayType cujo tipo de elemento seja também ArrayType, ao escrevê-lo em XML produz-se um nível de aninhamento extra que não está presente durante a conversão de ida e volta de ficheiros XML. Isto afeta apenas DataFrames provenientes de fora do XML — ler e escrever ficheiros XML preserva a estrutura original.

Por exemplo, um DataFrame com o seguinte esquema:

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

e os seguintes dados:

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

produz a seguinte saída XML:

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

O nome do elemento da matriz sem nome no DataFrame é especificado pela opção arrayElementName (Padrão: item).

Ativar a coluna de dados resgatados

A coluna de dados recuperados assegura que nunca se perdem dados durante o processo de ETL. Captura quaisquer dados que não tenham sido analisados porque um ou mais campos num registo apresentam um dos seguintes problemas:

  • Ausente do esquema fornecido.
  • Não corresponde ao tipo de dados do esquema fornecido.
  • Há uma incompatibilidade de maiúsculas e minúsculas nos nomes de campo do esquema fornecido.

A coluna de dados resgatados é retornada como um documento JSON contendo as colunas que foram resgatadas e o caminho do arquivo de origem do registro.

Para ativar a coluna de dados recuperados, defina a opção rescuedDataColumn com um nome de coluna durante a leitura:

Python

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

linguagem de programação 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'
)

Para remover o caminho do ficheiro de origem da coluna de dados resgatados, defina:

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

O analisador XML suporta três modos ao analisar registros: PERMISSIVE, DROPMALFORMEDe FAILFAST. Quando utilizado em conjunto com rescuedDataColumn, as incompatibilidades de tipos de dados não fazem com que os registos sejam descartados no modo DROPMALFORMED nem geram um erro no modo FAILFAST. Somente registros corrompidos (XML incompleto ou malformado) são descartados ou geram erros.

Inferir e evoluir o esquema com o Auto Loader

Para obter uma discussão detalhada deste tópico e das opções aplicáveis, consulte Configurar inferência e evolução de esquema no Auto Loader. Você pode configurar o Auto Loader para detetar automaticamente o esquema de dados XML carregados, permitindo inicializar tabelas sem declarar explicitamente o esquema de dados e evoluir o esquema de tabela à medida que novas colunas são introduzidas. Isso elimina a necessidade de rastrear e aplicar manualmente as alterações de esquema ao longo do tempo.

Por padrão, a inferência de esquema do Auto Loader procura evitar problemas de evolução do esquema devido a incompatibilidades de tipo. Para formatos que não codificam tipos de dados (JSON, CSV e XML), o Auto Loader infere todas as colunas como cadeias de caracteres, incluindo campos aninhados em arquivos XML. O Apache Spark DataFrameReader usa um comportamento diferente para inferência de esquema, selecionando tipos de dados para colunas em fontes XML com base em dados de exemplo. Para habilitar esse comportamento com o Auto Loader, defina a opção cloudFiles.inferColumnTypes como true.

O Auto Loader deteta a adição de novas colunas à medida que processa os seus dados. Quando o Auto Loader deteta uma nova coluna, o fluxo para com um UnknownFieldException. Antes de o fluxo lançar este erro, o Auto Loader executa a inferência de esquema no mais recente microlote de dados e atualiza o local do esquema com o esquema mais recente, adicionando novas colunas no final do esquema. Os tipos de dados das colunas existentes permanecem inalterados. Auto Loader suporta diferentes modos de para evolução de esquema, que você define na opção cloudFiles.schemaEvolutionMode.

Você pode usar indicações de esquema para aplicar as informações de esquema que você conhece e espera em um esquema inferido. Quando você souber que uma coluna é de um tipo de dados específico, ou se quiser escolher um tipo de dados mais geral (por exemplo, um duplo em vez de um inteiro), poderá fornecer um número arbitrário de dicas para tipos de dados de coluna como uma cadeia de caracteres usando a sintaxe de especificação do esquema SQL. Quando a coluna de dados resgatados é ativada, os campos nomeados com um formato diferente do do esquema são carregados na coluna _rescued_data. Você pode alterar esse comportamento definindo a opção readerCaseSensitive como false, caso em que o Auto Loader lê dados de forma que não diferencia maiúsculas de minúsculas.

Usage

Os exemplos seguintes utilizam o conjunto de dados Wanderbricks para demonstrar a leitura e escrita de ficheiros XML usando a API Spark DataFrame e SQL.

Ler e escrever XML

Utilize a API do DataFrame para escrever avaliações do Wanderbricks em XML e voltar a lê-las.

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

linguagem de programação 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")

Você pode especificar manualmente o esquema ao ler dados:

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

linguagem de programação 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")

Leia e escreva XML com SQL

Usa SQL DDL para criar uma tabela a partir de um ficheiro XML. Azure Databricks infere automaticamente os tipos de colunas.

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

Você também pode especificar nomes e tipos de colunas em DDL. Neste caso, o esquema não é inferido automaticamente.

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

Carregar XML usando COPY INTO

Use COPY INTO para carregar ficheiros XML a partir de armazenamento na nuvem para uma tabela 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');

Ler XML com validação por linha

Utilize a opção rowValidationXSDPath para validar cada linha em relação a um esquema XSD durante a leitura.

Python

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

linguagem de programação 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'
)

Carregar XML com o carregador automático

Use o Auto Loader para ingerir continuamente ficheiros XML do armazenamento na cloud para uma tabela Delta com inferência e evolução automática de esquemas.

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

linguagem de programação 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")

Recursos adicionais

  • Last updated on