Ler e gravar arquivos do Avro

O Apache Avro é um formato de serialização de dados baseado em linha que fornece estruturas de dados avançadas e uma codificação binária compacta e rápida. Os usuários do Azure Databricks costumam se deparar com isso ao ingerir dados de sistemas de streaming de eventos, como Apache Kafka e Google Pub/Sub, em que o Avro é o formato de serialização dominante. Azure Databricks dá suporte ao Avro para leitura e gravação com o Apache Spark, incluindo conversão automática de esquema entre tipos SQL do Avro e Spark, particionamento, compactação e nomes de registro personalizados.

Se você estiver lendo registros codificados em Avro no Apache Kafka ou em outro barramento de mensagens, em vez de arquivos, consulte Ler e gravar dados Avro de streaming, que aborda as funções from_avro e to_avro usadas para desserialização de streaming.

Pré-requisitos

Azure Databricks não requer configuração adicional para usar arquivos Avro. No entanto, para transmitir arquivos do Avro, você precisa do Carregador Automático.

Opções

Use os métodos .option() e .options() de DataFrameReader e DataFrameWriter para configurar fontes de dados Avro. Para obter uma lista completa de opções com suporte, consulte DataFrameReader as opções do Avro e DataFrameWriter as opções do Avro.

Usage

Os exemplos a seguir usam o conjunto de dados do Wanderbricks para demonstrar a leitura e gravação de arquivos do Avro usando a API do DataFrame do Spark e o SQL.

Ler arquivos do Avro usando SQL

Para consultar arquivos do Avro sem registrar uma tabela, use read_files. As permissões do Catálogo do Unity no local externo se aplicam automaticamente.

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

Ler e gravar arquivos do Avro

Use a API de DataFrame do Apache Spark quando precisar ler ou gravar arquivos Avro para um sistema downstream, aplicar transformações antes de carregar ou controlar opções como particionamento e esquema no momento da gravação.

Os exemplos a seguir usam o conjunto de dados de exemplo do 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 (linguagem de programação)

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;

Recursos adicionais

  • Ler e gravar arquivos Parquet: se sua carga de trabalho for principalmente analítica e de leitura pesada em vez de streaming ou gravação pesada, o layout de coluna do Parquet oferecerá um desempenho de consulta mais eficiente do que o armazenamento baseado em linha da Avro.
  • Last updated on