Membaca dan menulis file Avro

Apache Avro adalah format serialisasi data berbasis baris yang menyediakan struktur data yang kaya dan pengodean biner yang ringkas dan cepat. Pengguna Azure Databricks paling sering menemukannya saat mengimpor data dari sistem streaming data peristiwa seperti Apache Kafka dan Google Pub/Sub, dengan Avro sebagai format serialisasi yang dominan. Azure Databricks mendukung Avro untuk membaca dan menulis dengan Apache Spark, termasuk konversi skema otomatis antara jenis Avro dan Spark SQL, partisi, pemadatan, dan nama rekaman kustom.

Jika Anda membaca record berkode Avro dari Apache Kafka atau bus pesan lainnya alih-alih dari file, lihat Membaca dan menulis data streaming Avro, yang membahas fungsi from_avro dan to_avro yang digunakan untuk deserialisasi streaming.

Prerequisites

Azure Databricks tidak memerlukan konfigurasi tambahan untuk menggunakan file Avro. Namun, untuk melakukan streaming file Avro, Anda memerlukan Auto Loader.

Opsi

.option() Gunakan metode .options() dan DataFrameReader dan DataFrameWriter untuk mengonfigurasi sumber data Avro. Untuk daftar lengkap opsi yang didukung, lihat DataFrameReader Opsi Avro dan DataFrameWriter opsi Avro.

Usage

Contoh berikut menggunakan himpunan data Wanderbricks untuk menunjukkan pembacaan dan penulisan file Avro menggunakan Spark DataFrame API dan SQL.

Membaca file Avro menggunakan SQL

Untuk mengkueri file Avro tanpa mendaftarkan tabel, gunakan read_files. Izin Katalog Unity pada lokasi eksternal berlaku secara otomatis.

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

Membaca dan menulis file Avro

Gunakan APACHE Spark DataFrame API saat Anda perlu membaca atau menulis file Avro untuk sistem hilir, menerapkan transformasi sebelum memuat, atau mengontrol opsi seperti partisi dan skema pada waktu tulis.

Contoh berikut menggunakan himpunan data sampel Wanderbricks .

Phyton

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.{col, month, year}

// 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;

Sumber daya tambahan

• Membaca dan menulis file Parquet: Jika beban kerja Anda terutama bersifat analitis dan lebih banyak melibatkan operasi baca daripada streaming atau operasi tulis, format kolumnar Parquet menawarkan kinerja kueri yang lebih efisien daripada penyimpanan berbasis baris Avro.