Avro-bestanden lezen en schrijven

Apache Avro is een op rijen gebaseerde indeling voor gegevensserialisatie die uitgebreide gegevensstructuren en een compacte, snelle binaire codering biedt. Azure Databricks-gebruikers komen dit het vaakst tegen bij het opnemen van data uit eventstreamingsystemen zoals Apache Kafka en Google Pub/Sub, waarbij Avro het dominante serialisatieformaat is. Azure Databricks ondersteunt Avro voor zowel lezen als schrijven met Apache Spark, waaronder automatische schemaconversie tussen Avro- en Spark SQL-typen, partitionering, compressie en aangepaste recordnamen.

Als u Avro-gecodeerde records uit Apache Kafka of een andere berichtenbus leest in plaats van uit bestanden, zie Streaming-Avrogegevens lezen en schrijven, waarin de functies from_avro en to_avro voor deserialisatie van streams worden behandeld.

Vereiste voorwaarden

Azure Databricks vereist geen aanvullende configuratie voor het gebruik van Avro-bestanden. Als u Avro-bestanden wilt streamen, hebt u echter automatische laadprogramma's nodig.

Options

Gebruik de .option() en .options() methoden van DataFrameReader en DataFrameWriter om Avro-gegevensbronnen te configureren. Zie Avro-opties en DataFrameReader Avro-opties voor een volledige lijst met ondersteunde optiesDataFrameWriter.

Usage

In de volgende voorbeelden wordt de Wanderbricks-gegevensset gebruikt om avro-bestanden te lezen en te schrijven met behulp van de Spark DataFrame-API en SQL.

Avro-bestanden lezen met SQL

Als u een query wilt uitvoeren op Avro-bestanden zonder een tabel te registreren, gebruikt u read_files. Unity Catalog-machtigingen op de externe locatie worden automatisch toegepast.

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

Avro-bestanden lezen en schrijven

Gebruik de Apache Spark DataFrame-API wanneer u Avro-bestanden wilt lezen of schrijven voor een downstreamsysteem, transformaties wilt toepassen voordat u laadt of besturingselementopties zoals partitioneren en schema tijdens het schrijven.

In de volgende voorbeelden wordt de wanderbricks-voorbeeldgegevensset gebruikt.

Python

from pyspark.sql.functions import year, month

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

# Read an Avro file into a DataFrame
df = spark.read.format("avro").load("/Volumes/<catalog>/<schema>/<volume>/reviews_avro")
display(df)

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

# Read using a custom Avro schema to select specific fields
avro_schema = """
{
  "type": "record",
  "name": "Review",
  "fields": [
    {"name": "review_id", "type": "string"},
    {"name": "rating", "type": "int"},
    {"name": "comment", "type": ["null", "string"]}
  ]
}
"""
df = spark.read.format("avro").option("avroSchema", avro_schema).load("/Volumes/<catalog>/<schema>/<volume>/reviews_avro")

# Write partitioned Avro files by year and 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("avro").partitionBy("year", "month").save("/Volumes/<catalog>/<schema>/<volume>/bookings_avro_partitioned")

# Write with a custom record name and namespace for Schema Registry compatibility
df = spark.read.table("samples.wanderbricks.reviews")
df.write.format("avro").options(
  recordName="Review",
  recordNamespace="com.wanderbricks"
).save("/Volumes/<catalog>/<schema>/<volume>/reviews_avro")

Scala

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

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

// Read an Avro file into a DataFrame
val df = spark.read.format("avro").load("/Volumes/<catalog>/<schema>/<volume>/reviews_avro")
df.show()

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

// Read using a custom Avro schema to select specific fields
val avroSchema = """
{
  "type": "record",
  "name": "Review",
  "fields": [
    {"name": "review_id", "type": "string"},
    {"name": "rating", "type": "int"},
    {"name": "comment", "type": ["null", "string"]}
  ]
}
"""
val filtered = spark.read.format("avro").option("avroSchema", avroSchema).load("/Volumes/<catalog>/<schema>/<volume>/reviews_avro")

// Write partitioned Avro files by year and 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("avro").partitionBy("year", "month").save("/Volumes/<catalog>/<schema>/<volume>/bookings_avro_partitioned")

// Write with a custom record name and namespace for Schema Registry compatibility
reviews.write.format("avro").options(Map(
  "recordName" -> "Review",
  "recordNamespace" -> "com.wanderbricks"
)).save("/Volumes/<catalog>/<schema>/<volume>/reviews_avro")

SQL

-- Write wanderbricks reviews to Avro format
CREATE TABLE reviews_avro
USING AVRO
AS SELECT * FROM samples.wanderbricks.reviews;

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

SELECT * FROM bookings_avro_partitioned;

Aanvullende bronnen

• Parquet-bestanden lezen en schrijven: als uw workload voornamelijk analytische en leesintensieve is in plaats van streaming of schrijfintensief, biedt de kolomindeling van Parquet efficiëntere queryprestaties dan de opslag op basis van rijen van Avro.