Lire et écrire des fichiers Parquet

Apache Parquet est un format de fichier columnar optimisé pour les charges de travail analytiques. Il permet aux moteurs de requête de lire uniquement les colonnes nécessaires et d’ignorer les groupes de lignes non pertinents. Parquet est le format de stockage sous-jacent pour Delta Lake(/delta/index.md), ce qui en fait le format le plus courant pour les données stockées dans Azure Databricks. Azure Databricks prend en charge Parquet pour la lecture et l’écriture avec Apache Spark, notamment la spécification du schéma, le partitionnement et la compression d’écriture.

Prerequisites

Azure Databricks ne nécessite pas de configuration supplémentaire pour utiliser des fichiers Parquet. Toutefois, pour diffuser en continu des fichiers Parquet, vous avez besoin d’un chargeur automatique.

Paramètres

Utilisez les méthodes .option() et .options() de DataFrameReader et DataFrameWriter pour configurer des sources de données Parquet. Pour obtenir la liste complète des options prises en charge, consultez DataFrameReader Options Parquet et DataFrameWriter Options Parquet.

Usage

Les exemples suivants utilisent l’exemple de jeu de données Wanderbricks pour illustrer la lecture et l’écriture de fichiers Parquet à l’aide de l’API DataFrame Spark et de SQL.

Lire des fichiers Parquet à l’aide de SQL

Permet read_files d’interroger des fichiers Parquet directement à partir du stockage cloud à l’aide de SQL sans créer de table.

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews_parquet',
  format => 'parquet'
)

Lire et écrire des fichiers Parquet

Les exemples suivants écrivent les avis Wanderbricks au format Parquet, les relisent dans un DataFrame et illustrent le mode de remplacement.

Python

# Write wanderbricks reviews to Parquet format
df = spark.read.table("samples.wanderbricks.reviews")
df.write.format("parquet").save("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet")

# Read a Parquet file into a DataFrame
df = spark.read.format("parquet").load("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet")
display(df)

# Write with overwrite mode
df.write.format("parquet").mode("overwrite").save("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet")

Scala

// Write wanderbricks reviews to Parquet format
val reviews = spark.read.table("samples.wanderbricks.reviews")
reviews.write.format("parquet").save("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet")

// Read a Parquet file into a DataFrame
val df = spark.read.format("parquet").load("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet")
df.show()

// Write with overwrite mode
df.write.format("parquet").mode("overwrite").save("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet")

SQL

-- Write wanderbricks reviews to Parquet format
CREATE TABLE reviews_parquet
USING PARQUET
AS SELECT * FROM samples.wanderbricks.reviews;

SELECT * FROM reviews_parquet;

Spécifier un schéma

Spécifiez un schéma lors de la lecture de fichiers Parquet pour éviter la surcharge de l’inférence de schéma. Par exemple, définissez un schéma avec les champs review_id, rating et comment, puis lisez reviews_parquet dans un DataFrame.

Python

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("review_id", StringType(), True),
    StructField("rating", IntegerType(), True),
    StructField("comment", StringType(), True)
])

df = spark.read.format("parquet").schema(schema).load("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet")
df.printSchema()
df.show()

Scala

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

val schema = StructType(Array(
  StructField("review_id", StringType, nullable = true),
  StructField("rating", IntegerType, nullable = true),
  StructField("comment", StringType, nullable = true)
))

val df = spark.read.format("parquet").schema(schema).load("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet")
df.printSchema()
df.show()

SQL

-- Create a table with an explicit schema from Parquet files
CREATE TABLE reviews_parquet (
  review_id STRING,
  rating INT,
  comment STRING
)
USING PARQUET
OPTIONS (path "/Volumes/<catalog>/<schema>/<volume>/reviews_parquet");

SELECT * FROM reviews_parquet;

Écrire des fichiers Parquet partitionnés

Écrivez des fichiers Parquet partitionnés pour optimiser les performances des requêtes sur des jeux de données volumineux. Par exemple, lisez samples.wanderbricks.bookings et écrivez-le vers bookings_parquet_partitioned, partitionné par year et month, dérivés de la colonne check_in.

Python

from pyspark.sql.functions import year, month

df = spark.read.table("samples.wanderbricks.bookings")
df_with_parts = df.withColumn("year", year("check_in")).withColumn("month", month("check_in"))
df_with_parts.write.format("parquet").partitionBy("year", "month").save("/Volumes/<catalog>/<schema>/<volume>/bookings_parquet_partitioned")

Scala

import org.apache.spark.sql.functions.{col, month, year}

val bookings = spark.read.table("samples.wanderbricks.bookings")
val bookingsWithParts = bookings.withColumn("year", year(col("check_in"))).withColumn("month", month(col("check_in")))
bookingsWithParts.write.format("parquet").partitionBy("year", "month").save("/Volumes/<catalog>/<schema>/<volume>/bookings_parquet_partitioned")

SQL

-- Write partitioned Parquet files by year and month
CREATE TABLE bookings_parquet_partitioned
USING PARQUET
PARTITIONED BY (year, month)
AS SELECT *, year(check_in) AS year, month(check_in) AS month
FROM samples.wanderbricks.bookings;

Ressources additionnelles

• Qu'est-ce que Delta Lake dans Azure Databricks ? : si vous avez besoin de transactions ACID, d'une application de schéma ou d'un voyage temporel en même temps que les performances en colonnes de Parquet, Delta Lake est le format recommandé pour les données stockées dans Azure Databricks.