Lesen und Schreiben von Avro-Dateien

Apache Avro ist ein zeilenbasiertes Daten serialisierungsformat, das umfangreiche Datenstrukturen und eine kompakte, schnelle binäre Codierung bereitstellt. Azure-Databricks-Benutzer begegnen ihm am häufigsten bei der Aufnahme von Daten aus Event-Streaming-Systemen wie Apache Kafka und Google Pub/Sub, wobei Avro das dominierende Serialisierungsformat ist. Azure Databricks unterstützt Avro für das Lesen und Schreiben mit Apache Spark, einschließlich der automatischen Schemakonvertierung zwischen Avro- und Spark SQL-Typen, Partitionierung, Komprimierung und benutzerdefinierten Datensatznamen.

Wenn Sie Avro-kodierte Datensätze aus Apache Kafka oder einem anderen Nachrichtenbus statt aus Dateien lesen, lesen Sie Streaming-Avro-Daten lesen und schreiben. Dort werden die für die Streaming-Deserialisierung verwendeten from_avro- und to_avro-Funktionen beschrieben.

Voraussetzungen

Azure Databricks erfordert keine zusätzliche Konfiguration für die Verwendung von Avro-Dateien. Um avro-Dateien zu streamen, benötigen Sie jedoch das automatische Laden.

Options

Verwenden Sie die Methoden .option() und .options() von DataFrameReader und DataFrameWriter, um Avro-Datenquellen zu konfigurieren. Eine vollständige Liste der unterstützten Optionen finden Sie unter DataFrameReader Avro-Optionen und DataFrameWriter Avro-Optionen.

Usage

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

Lesen von Avro-Dateien mit SQL

Wenn Sie Avro-Dateien abfragen möchten, ohne eine Tabelle zu registrieren, verwenden Sie read_files. Unity-Katalogberechtigungen für den externen Speicherort werden automatisch angewendet.

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

Lesen und Schreiben von Avro-Dateien

Verwenden Sie die Apache Spark DataFrame-API, wenn Sie Avro-Dateien für ein nachgeschaltetes System lesen oder schreiben müssen, Transformationen vor dem Laden anwenden oder Steuerungsoptionen wie Partitionierung und Schema zum Zeitpunkt der Schreibzeit.

In den folgenden Beispielen wird der Wanderbricks-Beispieldatensatz verwendet.

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;

Weitere Ressourcen

• Parquet-Dateien lesen und schreiben: Wenn Ihre Workload in erster Linie analytisch und leselastig ist und nicht auf Streaming oder schreibintensive Vorgänge ausgerichtet ist, bietet das spaltenbasierte Layout von Parquet eine effizientere Abfrageperformance als die zeilenbasierte Speicherung von Avro.