Parquet-Dateien lesen und schreiben

Apache Parquet ist ein spaltenorientiertes Dateiformat, das für analytische Workloads optimiert ist. Es ermöglicht Abfragemodulen, nur die benötigten Spalten zu lesen und irrelevante Zeilengruppen zu überspringen. Parkett ist das zugrunde liegende Speicherformat für Delta Lake(/delta/index.md), wodurch es das am häufigsten verwendete Format für Daten ist, die in Azure Databricks gespeichert sind. Azure Databricks unterstützt Parquet zum Lesen und Schreiben mit Apache Spark, einschließlich Schemadefinition, Partitionierung und Komprimierung beim Schreiben.

Voraussetzungen

Azure Databricks erfordert keine zusätzliche Konfiguration für die Verwendung von Parkettdateien. Um jedoch Parkettdateien zu streamen, benötigen Sie auto Loader.

Optionen

Verwenden Sie die Methoden .option() und .options() von DataFrameReader und DataFrameWriter zum Konfigurieren von Parquet-Datenquellen. Eine vollständige Liste der unterstützten Optionen finden Sie unter DataFrameReader Parkettoptionen und DataFrameWriter Parkettoptionen.

Usage

In den folgenden Beispielen wird das Wanderbricks-Beispiel-Dataset verwendet, um das Lesen und Schreiben von Parkettdateien mithilfe der Spark DataFrame-API und SQL zu veranschaulichen.

Parquet-Dateien mit SQL lesen

Wird read_files verwendet, um Parkettdateien direkt aus dem Cloudspeicher mithilfe von SQL abzufragen, ohne eine Tabelle zu erstellen.

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

Parquet-Dateien lesen und schreiben

Die folgenden Beispiele schreiben die Wanderbricks-Bewertungen in das Parquet-Format, lesen sie wieder in einen DataFrame ein und demonstrieren den Überschreibungsmodus.

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;

Angeben eines Schemas

Geben Sie beim Lesen von Parquet-Dateien ein Schema an, um den Mehraufwand der Schemainferenz zu vermeiden. Definieren Sie z. B. ein Schema mit den Feldern review_id, rating und comment und lesen Sie reviews_parquet in einen DataFrame ein.

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;

Partitionierte Parkettdateien schreiben

Schreiben Sie partitionierte Parquet-Dateien für eine optimierte Abfrageleistung bei großen Datensätzen. Lesen Sie beispielsweise samples.wanderbricks.bookings und schreiben Sie es nach bookings_parquet_partitioned, partitioniert nach year und month, die aus der Spalte check_in abgeleitet werden.

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.{year, month}

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;

Weitere Ressourcen

• Was ist Delta Lake in Azure Databricks?: Wenn Sie ACID-Transaktionen, Schemaerzwingung oder Time Travel zusammen mit der spaltenorientierten Leistung von Parquet benötigen, ist Delta Lake das empfohlene Format für in Azure Databricks gespeicherte Daten.