Lire et écrire des données XML à l’aide de la bibliothèque spark-xml

Cet article explique comment lire et écrire un fichier XML en tant que Apache Spark source de données.

Spécifications

1. Créez la spark-xml bibliothèque en tant que bibliothèque Maven. Pour la coordonnée Maven, spécifiez :

   • Databricks Runtime 7.x et ultérieur : com.databricks:spark-xml_2.12:<release>

   Consultez spark-xmlVersions pour connaître la dernière version de <release>.

2. Installer la bibliothèque sur un cluster.

Exemple

L’exemple de cette section utilise le fichier XML books .

1. Récupérez le fichier XML des livres :

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

2. Télécharger le fichier à DBFS.

Lire et écrire des données 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")

Langage de programmation 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")

Paramètres

• Lire
  • path: Emplacement des fichiers XML. Accepte les expressions globbing Hadoop standard.
  • rowTag: Balise de ligne à traiter en tant que ligne. Par exemple, dans ce code XML <books><book><book>...</books>, la valeur est book. La valeur par défaut est ROW.
  • samplingRatio: Taux d’échantillonnage pour la déduction du schéma (0,0 ~ 1). 1 constitue la valeur par défaut. Les types possibles sont StructType , ArrayType , StringType , LongType , DoubleType , BooleanType, TimestampType et NullType , sauf si vous fournissez un schéma.
  • excludeAttribute: Indique s’il faut exclure des attributs dans les éléments. La valeur par défaut est false.
  • nullValue: Valeur à traiter en tant que valeur null. La valeur par défaut est "".
  • mode: Mode de traitement des enregistrements endommagés. La valeur par défaut est PERMISSIVE.
    • PERMISSIVE :
      • Lorsqu'il rencontre un enregistrement endommagé, il attribue la valeur null à tous les champs et place la chaîne incorrecte dans un nouveau champ configuré par columnNameOfCorruptRecord.
      • Lorsqu’il rencontre un champ de type de données incorrect, définit le champ incriminé sur null .
    • DROPMALFORMED: ignore les enregistrements endommagés.
    • FAILFAST : lève une exception en présence d’enregistrements endommagés.
  • inferSchema: si true , tente de déduire un type approprié pour chaque colonne tableau résultante, comme un type booléen, numérique ou de date. Si false la valeur est, toutes les colonnes résultantes sont de type chaîne. La valeur par défaut est true.
  • columnNameOfCorruptRecord: Nom du nouveau champ dans lequel sont stockées les chaînes mal formées. La valeur par défaut est _corrupt_record.
  • attributePrefix: Préfixe des attributs afin de différencier les attributs et les éléments. Il s’agit du préfixe pour les noms de champs. La valeur par défaut est _.
  • valueTag: Balise utilisée pour la valeur lorsqu’il existe des attributs dans un élément qui n’a pas d’éléments enfants. La valeur par défaut est _VALUE.
  • charset : Par défaut sur UTF-8, mais peut être défini sur d’autres noms de jeux de caractères valides.
  • ignoreSurroundingSpaces: Indique si les espaces blancs qui entourent les valeurs doivent être ignorés. La valeur par défaut est false.
  • rowValidationXSDPath: Chemin d’accès à un fichier XSD utilisé pour valider le code XML de chaque ligne. Les lignes qui ne peuvent pas être validées sont traitées comme des erreurs d’analyse comme indiqué ci-dessus. Le XSD n’affecte pas le schéma fourni ou inféré. Si le même chemin d’accès local n’est pas déjà également visible sur les exécuteurs du cluster, le schéma XSD et les autres éléments dont il dépend doivent être ajoutés aux exécuteurs Spark avec SparkContext.addFile. Dans ce cas, pour utiliser le XSD local /foo/bar.xsd, appelez addFile("/foo/bar.xsd") et transmettez "bar.xsd" en tant que rowValidationXSDPath .
• Écrire
  • path: Emplacement d’écriture des fichiers.
  • rowTag: Balise de ligne à traiter en tant que ligne. Par exemple, dans ce code XML <books><book><book>...</books>, la valeur est book. La valeur par défaut est ROW.
  • rootTag: Balise racine à traiter comme racine. Par exemple, dans ce code XML <books><book><book>...</books>, la valeur est books. La valeur par défaut est ROWS.
  • nullValue: La valeur à écrire pour null. La valeur par défaut est la chaîne "null" . Quand "null" est présent, il n’écrit pas les attributs et les éléments des champs.
  • attributePrefix: Le préfixe des attributs pour différencier les attributs des éléments. Il s’agit du préfixe pour les noms de champs. La valeur par défaut est _.
  • valueTag: Balise utilisée pour la valeur lorsqu’il existe des attributs dans un élément qui n’a pas d’éléments enfants. La valeur par défaut est _VALUE.
  • compression: Codec de compression à utiliser lors de l’enregistrement dans le fichier. Doit être le nom qualifié complet d’une classe qui implémente org.apache.hadoop.io.compress.CompressionCodec ou un des noms courts insensibles à la casse (bzip2, gzip, lz4 et snappy). La valeur par défaut est aucune compression.

Prend en charge l’utilisation raccourcie des noms ; Vous pouvez utiliser xml à la place de com.databricks.spark.xml .

Prise en charge de l'XSD

Vous pouvez valider des lignes individuelles par rapport à un schéma XSD à l’aide de rowValidationXSDPath.

Vous utilisez l'utilitaire com.databricks.spark.xml.util.XSDToSchema pour extraire un schéma de DataFrame Spark à partir de fichiers XSD. Il prend uniquement en charge les types simples, complexes et de séquence, uniquement les fonctionnalités XSD de base et est expérimental.

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

Analyser les données XML imbriquées

Bien qu’il soit principalement utilisé pour convertir un fichier XML en tableau, vous pouvez également utiliser la méthode from_xml pour analyser du code XML dans une colonne à valeur de chaîne d’un tableau existant et l'ajouter en tant que nouvelle colonne avec des résultats analysés en tant que struct avec :

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

Règles de conversion

En raison des différences structurelles entre les DataFrames et XML, il existe des règles de conversion des données XML vers des DataFrames et des DataFrames vers les données XML. Vous pouvez désactiver la gestion des attributs avec l’option excludeAttribute .

Convertir XML en tableau

• Attributs : les attributs sont convertis en champs avec le préfixe spécifié dans l'option attributePrefix. Si attributePrefix est _, le document

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

  produit le schéma :

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

• Si un élément possède des attributs mais pas d’éléments enfants, la valeur de l’attribut est placée dans un champ séparé spécifié dans l'option valueTag. Si valueTag est _VALUE, le document

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

  produit le schéma :

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

Convertir tableau en XML

L’écriture d’un fichier XML à partir d’un DataFrame, ayant un champ ArrayType avec son élément comme ArrayType, aurait un champ imbriqué supplémentaire pour l’élément. Cela ne se produit pas lors de la lecture et de l’écriture de données XML, mais lors de l’écriture d’un DataFrame qui a été lu à partir d’autres sources. Par conséquent, l'aller-retour lors de la lecture et de l'écriture de fichiers XML conserve la même structure, mais l'écriture d'un DataFrame lu à partir d'autres sources peut avoir une structure différente.

Un DataFrame avec le schéma :

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

et des données :

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

produit le fichier XML :

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