读取和写入 Avro 文件

Apache Avro 是一种基于行的数据序列化格式，可提供丰富的数据结构和紧凑、快速的二进制编码。 Azure Databricks用户最常在从 Apache Kafka 和 Google Pub/Sub 等事件流系统引入数据时遇到它，其中 Avro 是主要的序列化格式。 Azure Databricks支持 Avro 通过 Apache Spark 进行读取和写入，包括 Avro 和 Spark SQL 类型、分区、压缩和自定义记录名称之间的自动架构转换。

如果您读取的是来自 Apache Kafka 或其他消息总线而非文件的 Avro 编码记录，请参阅读取和写入流式 Avro 数据，其中介绍了用于流式反序列化的 from_avro 和 to_avro 函数。

Prerequisites

Azure Databricks不需要其他配置即可使用 Avro 文件。但是，若要流式传输 Avro 文件，需要自动加载程序。

选项

使用 .option() 和 .options() 的 DataFrameReader 和 DataFrameWriter 方法来配置 Avro 数据源。有关支持选项的完整列表，请参阅 DataFrameReader Avro 选项和 DataFrameWriter Avro 选项。

Usage

以下示例使用 Wanderbricks 数据集演示如何使用 Spark 数据帧 API 和 SQL 读取和写入 Avro 文件。

使用 SQL 读取 Avro 文件

若要在不注册表的情况下查询 Avro 文件，请使用 read_files。外部位置上的 Unity Catalog 权限会自动生效。

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

读取和写入 Avro 文件

如果需要读取或写入下游系统的 Avro 文件、在加载之前应用转换或控制选项（如在写入时分区和架构），请使用 Apache Spark 数据帧 API。

以下示例使用 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;

其他资源

读取和写入 Parquet 文件：如果工作负荷主要是分析和读取密集型文件，而不是流式处理或写入密集型，Parquet 的列式布局可提供比 Avro 基于行的存储更高效的查询性能。

Last updated on 2026-07-14