Ler e gravar dados XML usando a biblioteca spark-xml

Este artigo descreve como ler e gravar um arquivo XML como uma fonte de dados do Apache Spark.

Requisitos

1. Crie a biblioteca spark-xml como uma biblioteca do Maven. Para a coordenada do Maven, especifique:

   • Databricks Runtime 7.x e posterior: com.databricks:spark-xml_2.12:<release>

   Consulte spark-xmlVersões para obter a versão mais recente de <release>.

2. Instalar a biblioteca em um cluster.

Exemplo

O exemplo nesta seção usa o arquivo XML dos livros.

1. Recupere o arquivo XML dos livros:

   $ wget https://github.com/databricks/spark-xml/raw/master/src/test/resources/books.xml
   

2. Carregue o arquivo no DBFS.

Ler e gravar dados XML

SQL

/*Infer schema*/

CREATE TABLE books
USING xml
OPTIONS (path "dbfs:/books.xml", rowTag "book")

/*Specify column names and types*/

CREATE TABLE books (author string, description string, genre string, _id string, price double, publish_date string, title string)
USING xml
OPTIONS (path "dbfs:/books.xml", rowTag "book")

Scala (linguagem de programação)

// Infer schema

import com.databricks.spark.xml._ // Add the DataFrame.read.xml() method

val df = spark.read
  .option("rowTag", "book")
  .xml("dbfs:/books.xml")

val selectedData = df.select("author", "_id")
selectedData.write
  .option("rootTag", "books")
  .option("rowTag", "book")
  .xml("dbfs:/newbooks.xml")

// Specify schema

import org.apache.spark.sql.types.{StructType, StructField, StringType, DoubleType}

val customSchema = StructType(Array(
  StructField("_id", StringType, nullable = true),
  StructField("author", StringType, nullable = true),
  StructField("description", StringType, nullable = true),
  StructField("genre", StringType, nullable = true),
  StructField("price", DoubleType, nullable = true),
  StructField("publish_date", StringType, nullable = true),
  StructField("title", StringType, nullable = true)))

val df = spark.read
  .option("rowTag", "book")
  .schema(customSchema)
  .xml("books.xml")

val selectedData = df.select("author", "_id")
selectedData.write
  .option("rootTag", "books")
  .option("rowTag", "book")
  .xml("dbfs:/newbooks.xml")

R

# Infer schema

library(SparkR)

sparkR.session("local[4]", sparkPackages = c("com.databricks:spark-xml_2.12:<release>"))

df <- read.df("dbfs:/books.xml", source = "xml", rowTag = "book")

# Default `rootTag` and `rowTag`
write.df(df, "dbfs:/newbooks.xml", "xml")

# Specify schema

customSchema <- structType(
  structField("_id", "string"),
  structField("author", "string"),
  structField("description", "string"),
  structField("genre", "string"),
  structField("price", "double"),
  structField("publish_date", "string"),
  structField("title", "string"))

df <- read.df("dbfs:/books.xml", source = "xml", schema = customSchema, rowTag = "book")

# In this case, `rootTag` is set to "ROWS" and `rowTag` is set to "ROW".
write.df(df, "dbfs:/newbooks.xml", "xml", "overwrite")

Opções

• Ler
  • path: local dos arquivos XML. Aceita expressões de recurso de curinga padrão do Hadoop.
  • rowTag: a marca de linha a ser tratada como uma linha. Por exemplo, neste XML <books><book><book>...</books>, o valor seria book. O padrão é ROW.
  • samplingRatio: taxa de amostragem para inferir esquema (0,0 ~ 1). O padrão é UTF-1. Os tipos possíveis são StructType, ArrayType, StringType, LongType, DoubleType, BooleanType, TimestampType e NullType, a menos que você forneça um esquema.
  • excludeAttribute: se deve excluir atributos nos elementos. O padrão é false.
  • nullValue: o valor a ser tratado como um valor null. O padrão é "".
  • mode: o modo para lidar com registros corrompidos. O padrão é PERMISSIVE.
    • PERMISSIVE:
      • Quando encontra um registro corrompido, define todos os campos como null e coloca a cadeia de caracteres malformada em um novo campo configurado pelo columnNameOfCorruptRecord.
      • Quando encontra um campo do tipo de dados errado, define o campo incorreto como null.
    • DROPMALFORMED: ignora registros corrompidos.
    • FAILFAST: lança uma exceção quando detecta registros corrompidos.
  • inferSchema: se for true, tenta inferir um tipo apropriado para cada coluna do DataFrame resultante, como um tipo booliano, numérico ou de data. Se for false, todas as colunas resultantes serão do tipo de cadeia de caracteres. O padrão é true.
  • columnNameOfCorruptRecord: o nome do novo campo em que cadeias de caracteres malformadas são armazenadas. O padrão é _corrupt_record.
  • attributePrefix: o prefixo para atributos para diferenciar atributos e elementos. Esse é o prefixo para nomes de campo. O padrão é _.
  • valueTag: a marca usada para o valor quando há atributos em um elemento que não tem elementos filho. O padrão é _VALUE.
  • charset: o padrão é UTF-8, mas pode ser definido para outros nomes de conjunto de caracteres válidos.
  • ignoreSurroundingSpaces: se espaços em branco ao redor dos valores devem ou não ser ignorados. O padrão é false.
  • rowValidationXSDPath: caminho para um arquivo XSD usado para validar o XML para cada linha. As linhas que não são validadas são tratadas como erros de análise, conforme mostrado acima. O XSD não afeta o esquema fornecido ou inferido. Se o mesmo caminho local ainda não estiver visível nos executores no cluster, o XSD e quaisquer outros dos quais que ele dependa deverão ser adicionados aos executores do Spark com SparkContext.addFile. Nesse caso, para usar o XSD local /foo/bar.xsd, chame addFile("/foo/bar.xsd") e passe "bar.xsd" como rowValidationXSDPath.
• Gravar
  • path: local para gravar arquivos.
  • rowTag: a marca de linha a ser tratada como uma linha. Por exemplo, neste XML <books><book><book>...</books>, o valor seria book. O padrão é ROW.
  • rootTag: a marca raiz a ser tratada como a raiz. Por exemplo, neste XML <books><book><book>...</books>, o valor seria books. O padrão é ROWS.
  • nullValue: o valor para gravar o valor null. O padrão é a cadeia de caracteres "null". Quando "null", ele não grava atributos e elementos para os campos.
  • attributePrefix: o prefixo para atributos para diferenciar atributos e elementos. Esse é o prefixo para nomes de campo. O padrão é _.
  • valueTag: a marca usada para o valor quando há atributos em um elemento que não tem elementos filho. O padrão é _VALUE.
  • compression: codec de compactação a ser usado ao salvar no arquivo. Deve ser o nome totalmente qualificado de uma classe que implementa org.apache.hadoop.io.compress.CompressionCodec ou um dos nomes curtos que não sejam sensíveis a maiúsculas e minúsculas (bzip2, gzip, lz4 e snappy). O padrão é nenhuma compactação.

Dá suporte ao uso de nome reduzido; você pode usar em xml vez de com.databricks.spark.xml.

Suporte a XSD

Você pode validar linhas individuais em um esquema XSD usando rowValidationXSDPath.

Use o utilitário com.databricks.spark.xml.util.XSDToSchema para extrair um esquema do DataFrame do Spark de alguns arquivos XSD. Ele dá suporte apenas a tipos simples, complexos e de sequência, apenas à funcionalidade básica do XSD e é experimental.

import com.databricks.spark.xml.util.XSDToSchema
import java.nio.file.Paths

val schema = XSDToSchema.read(Paths.get("/path/to/your.xsd"))
val df = spark.read.schema(schema)....xml(...)

Analisar XML aninhado

Embora seja usado principalmente para converter um arquivo XML em um DataFrame, você também pode usar o método from_xml para analisar o XML em uma coluna com valor de cadeia de caracteres em um DataFrame existente e adicioná-lo como uma nova coluna com resultados analisados como um struct com:

import com.databricks.spark.xml.functions.from_xml
import com.databricks.spark.xml.schema_of_xml
import spark.implicits._

val df = ... /// DataFrame with XML in column 'payload'
val payloadSchema = schema_of_xml(df.select("payload").as[String])
val parsed = df.withColumn("parsed", from_xml($"payload", payloadSchema))

Regras de conversão

Devido a diferenças estruturais entre DataFrames e XML, há algumas regras de conversão de dados XML para DataFrame e de DataFrame para dados XML. Você pode desabilitar a manipulação de atributos com a opção excludeAttribute.

Converter XML em DataFrame

• Atributos: os atributos são convertidos como campos com o prefixo especificado na opção attributePrefix. Se attributePrefix for _, o documento

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

  produz o esquema:

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

• Se um elemento tiver atributos, mas não tiver nenhum elemento filho, o valor do atributo será colocado em um campo separado especificado na opção valueTag. Se valueTag for _VALUE, o documento

  <one>
      <two myTwoAttrib="BBBBB">two</two>
      <three>three</three>
  </one>
  

  produz o esquema:

  root
  |-- two: struct (nullable = true)
  |    |-- _VALUE: string (nullable = true)
  |    |-- _myTwoAttrib: string (nullable = true)
  |-- three: string (nullable = true)
  

Converter DataFrame em XML

A gravação de um arquivo XML do DataFrame que tenha um campo ArrayType com seu elemento definido como ArrayType teria um campo aninhado adicionado para o elemento. Isso não aconteceria na leitura e gravação de dados XML, mas na gravação de um DataFrame lido de outras fontes. Portanto, a viagem de ida e volta na leitura e gravação de arquivos XML tem a mesma estrutura, mas gravar um DataFrame lido de outras fontes é possível que tenha uma estrutura diferente.

Um DataFrame com o esquema:

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

e dados:

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

produz o arquivo XML:

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