Läs och skriv Parquet-filer

Apache Parquet är ett kolumnformat som är optimerat för analytiska arbetsbelastningar. Det gör att frågemotorer bara kan läsa de kolumner som behövs och hoppa över irrelevanta radgrupper. Parquet är det underliggande lagringsformatet för Delta Lake(/delta/index.md), vilket gör det till det vanligaste formatet för data som lagras i Azure Databricks. Azure Databricks stöder Parquet för både läsning och skrivning med Apache Spark, inklusive schemaspecifikation, partitionering och skrivkomprimering.

Förutsättningar

Azure Databricks kräver inte ytterligare konfiguration för att använda Parquet-filer. För att strömma Parquet-filer behöver du dock Auto Loader.

Alternativ

Använd metoderna .option() och .options() för DataFrameReader och DataFrameWriter för att konfigurera Parquet-datakällor. En fullständig lista över alternativ som stöds finns i DataFrameReader Parquet-alternativ och DataFrameWriter Parquet-alternativ.

Usage

I följande exempel används Wanderbricks-exempeldatamängden för att demonstrera läsning och skrivning av Parquet-filer med Spark DataFrame API och SQL.

Läs Parquet-filer med SQL

Använd read_files för att köra frågor mot Parquet-filer direkt från molnlagring med HJÄLP av SQL utan att skapa en tabell.

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

Läs och skriv Parquet-filer

I följande exempel skriver du Wanderbricks-recensionerna till Parquet-format, läser tillbaka dem till en DataFrame och demonstrerar överskrivningsläget.

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;

Ange ett schema

Ange ett schema när du läser Parquet-filer för att undvika omkostnaderna för schemainferens. Definiera till exempel ett schema med review_idfälten , ratingoch comment och läsa reviews_parquet in i en 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;

Skriva partitionerade Parquet-filer

Skriv partitionerade Parquet-filer för optimerad frågeprestanda på stora datauppsättningar. Till exempel, läs samples.wanderbricks.bookings och skriv till bookings_parquet_partitioned, partitionerad efter year och month som härleds från kolumnen 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;

Ytterligare resurser

• Vad är Delta Lake i Azure Databricks?: Om du behöver ACID-transaktioner, schemaframtvingande eller tidsresor tillsammans med Parquets kolumnprestanda är Delta Lake det rekommenderade formatet för data som lagras i Azure Databricks.