Archivo Avro

  • Artículo

Apache Avro es un sistema de serialización de datos. Avro proporciona:

  • Estructuras de datos enriquecidas.
  • Un formato de datos compacto, rápido y binario.
  • Un archivo contenedor para almacenar datos persistentes.
  • Llamada a procedimiento remoto (RPC).
  • Integración sencilla con lenguajes dinámicos. No se necesita generación de código para leer o escribir archivos de datos ni para usar o implementar protocolos RPC. La generación de código como una optimización opcional solo merece la pena implementarse para lenguajes con establecimiento estático de tipos.

El origen de datos Avro admite:

  • Conversión de esquemas: conversión automática entre Apache Spark SQL y registros Avro.
  • Creación de particiones: lectura y escritura sencillas de datos con particiones sin ninguna configuración adicional.
  • Compresión: compresión que se usa al escribir Avro en el disco. Los tipos admitidos son uncompressed, snappy y deflate. También puede especificar el nivel deflate.
  • Nombres de registro: nombre de registro y espacio de nombres al pasar un mapa de parámetros con recordName y recordNamespace.

Vea también Lectura y escritura de datos de Avro en streaming.

Configuración

Puede cambiar el comportamiento de un origen de datos de Avro mediante varios parámetros de configuración.

Para pasar por alto los archivos sin la extensión .avro al leer, puede establecer el parámetro avro.mapred.ignore.inputs.without.extension en la configuración de Hadoop. El valor predeterminado es false.

spark
  .sparkContext
  .hadoopConfiguration
  .set("avro.mapred.ignore.inputs.without.extension", "true")

Para configurar la compresión al escribir, establezca las siguientes propiedades de Spark:

  • Códec de compresión: spark.sql.avro.compression.codec. Los códecs admitidos son snappy y deflate. El códec predeterminado es snappy.
  • Si el códec de compresión es deflate, puede establecer el nivel de compresión con: spark.sql.avro.deflate.level. El nivel predeterminado es -1.

Puede establecer estas propiedades en la configuración de Spark del clúster o en tiempo de ejecución mediante spark.conf.set(). Por ejemplo:

spark.conf.set("spark.sql.avro.compression.codec", "deflate")
spark.conf.set("spark.sql.avro.deflate.level", "5")

En Databricks Runtime 9.1 LTS y posteriores, puede cambiar el comportamiento de inferencia de esquemas predeterminado de Avro si proporciona la opción mergeSchema al leer archivos. Al establecer mergeSchema en true, se infiere un esquema de un conjunto de archivos Avro del directorio de destino y se combinan en lugar de inferirse el esquema de lectura de un único archivo.

Tipos admitidos para Avro -> conversión de Spark SQL

Esta biblioteca admite la lectura de todos los tipos Avro. Usa la siguiente asignación de tipos Avro a tipos Spark SQL:

Tipo Avro Tipo Spark SQL
boolean BooleanType
int IntegerType
long LongType
FLOAT FloatType
double DoubleType
bytes BinaryType
string StringType
registro StructType
enum StringType
array ArrayType
map MapType
corregido BinaryType
union Vea Tipos de unión.

Tipos de unión

El origen de datos Avro admite tipos union de lectura. Avro considera que los tres tipos siguientes son tipos union:

  • union(int, long) se asigna a LongType.
  • union(float, double) se asigna a DoubleType.
  • union(something, null), donde something es cualquier tipo Avro compatible. Esto se asigna al mismo tipo Spark SQL que el de something, con nullable establecido en true.

Todos los demás tipos union son tipos complejos. Se asignan a StructType, donde los nombres de campo son member0, member1, y así sucesivamente, de acuerdo con los miembros de union. Esto es coherente con el comportamiento al convertir entre Avro y Parquet.

Tipos lógicos

El origen de datos Avro admite la lectura de los siguientes tipos lógicos de Avro:

Tipo lógico Avro Tipo Avro Tipo Spark SQL
date int DateType
timestamp-millis long TimestampType
timestamp-micros long TimestampType
Decimal corregido DecimalType
Decimal bytes DecimalType

Nota:

El origen de datos Avro omite los documentos, alias y otras propiedades presentes en el archivo Avro.

Tipos admitidos para la conversión de Spark SQL -> Avro

Esta biblioteca admite la escritura de todos los tipos Spark SQL en Avro. Para la mayoría de los tipos, la asignación de tipos de Spark a tipos Avro es sencilla (por ejemplo, IntegerType se convierte en int); la siguiente es una lista de los pocos casos especiales:

Tipo Spark SQL Tipo Avro Tipo lógico Avro
ByteType int
ShortType int
BinaryType bytes
DecimalType corregido Decimal
TimestampType long timestamp-micros
DateType int date

También puede especificar el esquema Avro de salida completo con la opción avroSchema, de modo que los tipos Spark SQL se puedan convertir en otros tipos Avro. Las conversiones siguientes no se aplican de manera predeterminada y requieren el esquema Avro especificado por el usuario:

Tipo Spark SQL Tipo Avro Tipo lógico Avro
ByteType corregido
StringType enum
DecimalType bytes Decimal
TimestampType long timestamp-millis

Ejemplos

En estos ejemplos se usa el archivo episodes.avro.

Scala

// The Avro records are converted to Spark types, filtered, and
// then written back out as Avro records

val df = spark.read.format("avro").load("/tmp/episodes.avro")
df.filter("doctor > 5").write.format("avro").save("/tmp/output")

En este ejemplo se muestra un esquema de Avro personalizado:

import org.apache.avro.Schema

val schema = new Schema.Parser().parse(new File("episode.avsc"))

spark
  .read
  .format("avro")
  .option("avroSchema", schema.toString)
  .load("/tmp/episodes.avro")
  .show()

En este ejemplo se muestran las opciones de compresión de Avro:

// configuration to use deflate compression
spark.conf.set("spark.sql.avro.compression.codec", "deflate")
spark.conf.set("spark.sql.avro.deflate.level", "5")

val df = spark.read.format("avro").load("/tmp/episodes.avro")

// writes out compressed Avro records
df.write.format("avro").save("/tmp/output")

En este ejemplo se muestran registros de Avro con particiones:

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().master("local").getOrCreate()

val df = spark.createDataFrame(
  Seq(
    (2012, 8, "Batman", 9.8),
    (2012, 8, "Hero", 8.7),
    (2012, 7, "Robot", 5.5),
    (2011, 7, "Git", 2.0))
  ).toDF("year", "month", "title", "rating")

df.toDF.write.format("avro").partitionBy("year", "month").save("/tmp/output")

En este ejemplo se muestran el nombre de registro y el espacio de nombres:

val df = spark.read.format("avro").load("/tmp/episodes.avro")

val name = "AvroTest"
val namespace = "org.foo"
val parameters = Map("recordName" -> name, "recordNamespace" -> namespace)

df.write.options(parameters).format("avro").save("/tmp/output")

Python

# Create a DataFrame from a specified directory
df = spark.read.format("avro").load("/tmp/episodes.avro")

#  Saves the subset of the Avro records read in
subset = df.where("doctor > 5")
subset.write.format("avro").save("/tmp/output")

SQL

Para consultar datos de Avro en SQL, registre el archivo de datos como una tabla o una vista temporal:

CREATE TEMPORARY VIEW episodes
USING avro
OPTIONS (path "/tmp/episodes.avro")

SELECT * from episodes

Ejemplo de cuaderno: lectura y escritura de archivos Avro

En el cuaderno siguiente se muestra cómo leer y escribir archivos Avro.

Lectura y escritura en el cuaderno de archivos Avro

Obtener el cuaderno