Ler e gravar dados XML usando a spark-xml biblioteca

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

Requisitos

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

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

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

2. Instale a biblioteca em um cluster.

Exemplo

O exemplo nesta seção usa o arquivo XML de 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")

linguagem de programação Scala

// 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: Localização dos ficheiros XML. Aceita expressões padrão de globbing do Hadoop.
  • rowTag: A etiqueta de linha a ser tratada como tal. Por exemplo, neste XML <books><book><book>...</books>, o valor seria book. A predefinição é ROW.
  • samplingRatio: Razão de amostragem para inferir esquema (0,0 ~ 1). A predefinição é 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 os atributos devem ser excluídos em elementos. O valor padrão é falso.
  • nullValue: O valor a ser tratado como um null valor. A predefinição é "".
  • mode: O modo para lidar com registros corrompidos. A predefinição é PERMISSIVE.
    • PERMISSIVE:
      • Quando ele encontra um registro corrompido, define todos os campos como null e coloca a cadeia de caracteres malformada em um novo campo configurado por columnNameOfCorruptRecord.
      • Quando encontrar um campo do tipo de dados errado, define o campo ofensivo como null.
    • DROPMALFORMED: ignora registros corrompidos.
    • FAILFAST: lança uma exceção quando deteta registros corrompidos.
  • inferSchema: se true, tenta inferir um tipo apropriado para cada coluna DataFrame resultante, como um tipo booleano, numérico ou de data. Se false, todas as colunas resultantes são do tipo cadeia de caracteres. A predefinição é true.
  • columnNameOfCorruptRecord: O nome do novo campo onde as cadeias de caracteres malformadas são armazenadas. A predefinição é _corrupt_record.
  • attributePrefix: O prefixo para atributos para diferenciar atributos e elementos. Este é o prefixo para nomes de campos. A predefinição é _.
  • valueTag: A tag usada para o valor quando há atributos em um elemento que não tem elementos filho. A predefinição é _VALUE.
  • charset: Por padrão, é UTF-8, mas pode ser definido para outros nomes de charsets válidos.
  • ignoreSurroundingSpaces: Se os espaços em branco em torno dos valores devem ou não ser ignorados. O valor padrão é falso.
  • rowValidationXSDPath: Caminho para um arquivo XSD que é usado para validar o XML para cada linha. As linhas que não validam são tratadas como erros de análise como acima. O XSD não afeta, de outra forma, 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 ele dependa devem ser adicionados aos executores do Spark com SparkContext.addFile. Nesse caso, para usar XSD /foo/bar.xsdlocal, chame addFile("/foo/bar.xsd") e passe "bar.xsd" como rowValidationXSDPath.
• Escrever
  • path: Local para gravar arquivos.
  • rowTag: A etiqueta de linha a ser tratada como tal. Por exemplo, neste XML <books><book><book>...</books>, o valor seria book. A predefinição é ROW.
  • rootTag: A etiqueta raiz que deve ser tratada como raiz. Por exemplo, neste XML <books><book><book>...</books>, o valor seria books. A predefinição é ROWS.
  • nullValue: O valor a ser gravado null . O padrão é a cadeia de caracteres "null". Quando "null", ele não escreve atributos e elementos para campos.
  • attributePrefix: O prefixo para atributos para diferenciar atributos e elementos. Este é o prefixo para nomes de campos. A predefinição é _.
  • valueTag: A tag usada para o valor quando há atributos em um elemento que não tem elementos filho. A predefinição é _VALUE.
  • compression: Codec de compressão para usar ao guardar num ficheiro. Deve ser o nome totalmente qualificado de uma classe implementando org.apache.hadoop.io.compress.CompressionCodec ou um dos nomes curtos que não diferenciam maiúsculas de minúsculas (bzip2, gzip, lz4, e snappy). O padrão é sem compressão.

Suporta o uso de nomes abreviados; Você pode usar xml em vez de com.databricks.spark.xml.

Suporte a XSD

Você pode validar linhas individuais em relação a um esquema XSD usando rowValidationXSDPath.

Use o utilitário com.databricks.spark.xml.util.XSDToSchema para extrair um esquema Spark DataFrame de alguns ficheiros XSD . Ele suporta apenas tipos simples, complexos e de sequência, apenas a funcionalidade XSD básica 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 usado principalmente para converter um arquivo XML em um DataFrame, você também pode usar o método from_xml para analisar 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 uma estrutura 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, existem algumas regras de conversão de dados XML para DataFrame e de DataFrame para dados XML. Você pode desativar 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 attributePrefix opção. Se attributePrefix estiver _, 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 nenhum elemento filho, o valor do atributo será colocado em um campo separado especificado na valueTag opção. Se valueTag estiver _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

Escrever um arquivo XML de DataFrame com um campo ArrayType com seu elemento como ArrayType teria um campo aninhado adicional para o elemento. Isso não aconteceria na leitura e gravação de dados XML, mas na escrita de um DataFrame lido de outras fontes. Portanto, a ida e volta ao ler e gravar ficheiros XML tem a mesma estrutura, mas ao escrever um DataFrame lido de outras fontes, é possível ter 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>