Avro ファイルの読み取りと書き込み

Apache Avro は、豊富なデータ構造とコンパクトで高速なバイナリ エンコードを提供する行ベースのデータシリアル化形式です。 Azure Databricks ユーザーがこれに最もよく遭遇するのは、Apache Kafka や Google Pub/Sub などのイベントストリーミングシステムからデータを取り込むときであり、そこでは Avro が主要なシリアライズ形式となっています。 Azure Databricksでは、Avro と Spark の SQL 型間の自動スキーマ変換、パーティション分割、圧縮、カスタム レコード名など、Apache Spark での読み取りと書き込みの両方に Avro がサポートされます。

ファイルからではなく Apache Kafka または別のメッセージ バスから Avro でエンコードされたレコードを読み取る場合は、「 ストリーミング Avro データの読み取りと書き込み」を参照してください。これは、逆シリアル化のストリーミングに使用される from_avro 関数と to_avro 関数について説明しています。

[前提条件]

Azure Databricks Avro ファイルを使用するために追加の構成は必要ありません。 ただし、Avro ファイルをストリーミングするには、 自動ローダーが必要です。

オプション

Avro データ ソースを構成するには、.option()と.options()のDataFrameReaderおよびDataFrameWriterメソッドを使用します。 サポートされているオプションの完全な一覧については、Avro オプションのDataFrameReaderと Avro オプションのDataFrameWriterを参照してください。

Usage

次の例では 、Wanderbricks データセット を使用して、Spark DataFrame API と SQL を使用した Avro ファイルの読み取りと書き込みを示します。

SQL を使用して Avro ファイルを読み取る

テーブルを登録せずに Avro ファイルにクエリを実行するには、 read_filesを使用します。 外部の場所に対する Unity カタログのアクセス許可は自動的に適用されます。

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

Avro ファイルの読み取りと書き込み

ダウンストリーム システムの Avro ファイルの読み取りまたは書き込み、読み込み前の変換の適用、または書き込み時のパーティション分割やスキーマなどのオプションの制御が必要な場合は、Apache Spark DataFrame 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")

スカラ (プログラミング言語)

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