Čtení a zápis souborů Avro

Apache Avro je formát serializace dat založený na řádcích, který poskytuje bohaté datové struktury a kompaktní a rychlé binární kódování. Azure Databricks se s nimi uživatelé nejčastěji setkávají při ingestování dat ze systémů streamování událostí, jako jsou Apache Kafka a Google Pub/Sub, kde Avro je dominantním formátem serializace. Azure Databricks podporuje Avro pro čtení i zápis pomocí Apache Sparku, včetně automatického převodu schématu mezi typy Avro a Spark SQL, dělením, kompresí a vlastními názvy záznamů.

Pokud čtete záznamy kódované ve formátu Avro z Apache Kafka nebo z jiné sběrnice zpráv, a ne ze souborů, viz Streamované čtení a zápis dat Avro, kde jsou popsány funkce from_avro a to_avro používané pro deserializaci datových proudů.

Předpoklady

Azure Databricks nevyžaduje další konfiguraci pro použití souborů Avro. Ke streamování souborů Avro však potřebujete Auto Loader.

Možnosti

Pomocí metod .option() a .options() objektů DataFrameReader a DataFrameWriter nakonfigurujte datové zdroje Avro. Úplný seznam podporovaných možností najdete v tématu DataFrameReader Možnosti Avro a DataFrameWriter Možnosti Avro.

Usage

Následující příklady používají datovou sadu Wanderbricks k předvedení čtení a zápisu souborů Avro pomocí rozhraní API datového rámce Sparku a SQL.

Čtení souborů Avro pomocí SQL

K dotazování souborů Avro bez registrace tabulky použijte read_files. Oprávnění Unity Catalog na externí umístění se použijí automaticky.

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

Čtení a zápis souborů Avro

Rozhraní DataFrame API v Apache Spark použijte, když potřebujete číst nebo zapisovat soubory Avro pro navazující systém, před načtením použít transformace nebo při zápisu nastavit možnosti, jako jsou particionování a schéma.

Následující příklady používají ukázkovou datovou sadu Wanderbricks .

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;

Dodatečné zdroje

• Čtení a zápis souborů Parquet: Pokud je vaše pracovní zátěž primárně analytická a zaměřená spíše na čtení než na streamování nebo častý zápis, sloupcové uspořádání formátu Parquet nabízí vyšší výkon při dotazování než řádkově orientované ukládání v Avro.