How to Encode and Decode Records with Avro

In this guide, we encode order events as Avro records with AvroFormat from ZIO Blocks Schema. By the end, you will have one codec, derived from a sealed trait, that writes the .avsc schema file another team needs, turns each event into compact bytes and back, writes and reads a stream of events, and produces bytes that Apache Avro's own Java library reads. A typical place for these bytes is a Kafka message body; the same bytes work in a file, a queue or an HTTP body, so nothing in this guide depends on a broker.

We start from the event types and build up in this order:

Derive the codec and write the .avsc schema file.
Encode and decode one event, then a stream of events.
See what AvroCodec#decode does with bad input.
Check how each Scala type maps to Avro, which a reader in another language must know.
Read and write the same bytes with Apache Avro.
Learn what the module leaves to you: schema evolution, mismatched schemas, container files and registry framing.

The Problem

Avro bytes hold only values. The writer and the reader must agree on a schema, a JSON document usually kept in an .avsc file, that says which field each value belongs to. When you write that schema by hand, it describes your case class a second time, and nothing keeps the two in step.

For example, consider the OrderShipped event. With Apache Avro alone, you write the schema as a JSON string and copy every field into a generic record by name:

// The schema repeats the case class by hand
val schemaJson =
  """{"type":"record","name":"OrderShipped","fields":[
    |  {"name":"orderId","type":"long"},
    |  {"name":"shippedOn","type":{"type":"record","name":"LocalDate","fields":[
    |    {"name":"year","type":"int"},{"name":"month","type":"int"},{"name":"day","type":"int"}]}},
    |  {"name":"carrier","type":"string"}]}""".stripMargin
val schema = new org.apache.avro.Schema.Parser().parse(schemaJson)

// Every field is copied by name: a misspelled name compiles, and fails only when this line runs
val shippedOn = new org.apache.avro.generic.GenericData.Record(schema.getField("shippedOn").schema())
shippedOn.put("year", 2024)
shippedOn.put("month", 3)
shippedOn.put("day", 9)

val record = new org.apache.avro.generic.GenericData.Record(schema)
record.put("orderId", 7L)
record.put("shippedOn", shippedOn)
record.put("carrier", "DHL")

// The writer and the encoder are set up by hand too
val out     = new java.io.ByteArrayOutputStream()
val encoder = org.apache.avro.io.EncoderFactory.get.binaryEncoder(out, null)
new org.apache.avro.generic.GenericDatumWriter[AnyRef](schema).write(record, encoder)
encoder.flush()

Add a field to the case class and you must remember to change the JSON, the GenericData.Record#put calls and the reading code in every service that consumes the event. A mistake shows up as a decoding failure in another service, often long after the change. In this guide, the case class is the single source: its Schema derives the Avro schema, the encoder and the decoder, so they cannot drift apart.

Prerequisites

Add the Avro module to your build. It runs on the JVM only; there is no Scala.js artifact:

// Scala 2.13 and Scala 3.3 to 3.8
libraryDependencies += "dev.zio" %% "zio-blocks-schema-avro" % "0.0.56"

// Scala 3.9 and newer (also needs scalacOptions += "-experimental")
libraryDependencies += "dev.zio" %% "zio-blocks-next-schema-avro" % "0.0.56"

The module depends on Apache Avro (org.apache.avro:avro), so the Apache Avro classes used later in this guide are already on the classpath. We assume you can derive a Schema for a case class, as shown in Your First Schema.

Every code block in this guide uses these imports:

import zio.blocks.schema._
import zio.blocks.schema.avro._
import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
import java.time.{Instant, LocalDate}

The Order Events

The running example is the life of an order: placed, shipped or cancelled. OrderEvent is a sealed trait and each event is a case class, so one codec covers all three:

sealed trait OrderEvent

object OrderEvent {
  implicit val schema: Schema[OrderEvent] = Schema.derived
}

case class OrderPlaced(
  orderId: Long,
  customer: String,
  amount: BigDecimal,
  placedAt: Instant,
  note: Option[String],
  items: List[LineItem]
) extends OrderEvent

case class OrderShipped(orderId: Long, shippedOn: LocalDate, carrier: String) extends OrderEvent

case class OrderCancelled(orderId: Long, reason: String) extends OrderEvent

case class LineItem(sku: String, quantity: Int, unitPrice: BigDecimal)

The fields cover what readers of Avro data usually ask about: ids, money, timestamps, an optional note and a list of line items.

The same model as a file, shared by every example: Model.scala

schema-examples/src/main/scala/avrorecords/Model.scala
package avrorecords

import zio.blocks.schema._
import java.time.{Instant, LocalDate}

sealed trait OrderEvent

object OrderEvent {
  implicit val schema: Schema[OrderEvent] = Schema.derived
}

case class OrderPlaced(
  orderId: Long,
  customer: String,
  amount: BigDecimal,
  placedAt: Instant,
  note: Option[String],
  items: List[LineItem]
) extends OrderEvent

case class OrderShipped(orderId: Long, shippedOn: LocalDate, carrier: String) extends OrderEvent

case class OrderCancelled(orderId: Long, reason: String) extends OrderEvent

case class LineItem(sku: String, quantity: Int, unitPrice: BigDecimal)

Write the .avsc Schema File

An Avro reader needs the schema that the writer encodes with. Derive the codec once from the schema of the sealed trait; the result is an AvroCodec[OrderEvent], which carries that schema:

val codec = OrderEvent.schema.derive(AvroFormat)

AvroCodec#avroSchema returns the schema as an Apache Avro Schema, and Apache Avro's SchemaFormatter turns it into the JSON text of an .avsc file. To print the schema of the smallest event, OrderCancelled:

import org.apache.avro.SchemaFormatter

println(SchemaFormatter.format("json/pretty", Schema.derived[OrderCancelled].derive(AvroFormat).avroSchema))
// {
//   "type" : "record",
//   "name" : "OrderCancelled",
//   "namespace" : "repl.MdocSession.MdocApp",
//   "fields" : [ {
//     "name" : "orderId",
//     "type" : "long"
//   }, {
//     "name" : "reason",
//     "type" : "string"
//   } ]
// }

The namespace above is repl.MdocSession.MdocApp because the documentation build compiles this page inside its own object. In your project, the namespace is the package of the case class (avrorecords in the example files), so moving a case class to another package changes the schema you share.

For the whole sealed trait, SchemaFormatter.format("json/pretty", codec.avroSchema) prints a union: a JSON array with one record per event, in the order the events are declared. Write that text to order-event.avsc with java.nio.file.Files#writeString and hand the file to the team that reads your events; the example file for this section does exactly that. SchemaFormatter is the printer since Apache Avro 1.12; the older Schema#toString(true) is deprecated.

Runnable version, which also writes order-event.avsc: Step1WriteAvscSchemaExample.scala

schema-examples/src/main/scala/avrorecords/Step1WriteAvscSchemaExample.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema-avro:0.0.56
//> using options -experimental
//> using file Model.scala
// Run from this folder:  scala-cli run Step1WriteAvscSchemaExample.scala
// Or from schema-examples:  sbt "runMain avrorecords.Step1WriteAvscSchemaExample"

package avrorecords

import org.apache.avro.SchemaFormatter
import zio.blocks.schema._
import zio.blocks.schema.avro._
import java.nio.file.{Files, Paths}

/**
 * How to Encode and Decode Records with Avro — Write the .avsc Schema File
 *
 * Derives the codec, prints the schema of `OrderCancelled`, and writes the
 * schema of the whole sealed trait to `order-event.avsc`.
 */
object Step1WriteAvscSchemaExample {
  def main(args: Array[String]): Unit = {
    val codec = OrderEvent.schema.derive(AvroFormat)

    println(SchemaFormatter.format("json/pretty", Schema.derived[OrderCancelled].derive(AvroFormat).avroSchema))

    // The whole sealed trait, written to order-event.avsc in the current directory
    val avscText = SchemaFormatter.format("json/pretty", codec.avroSchema)
    val avscFile = Paths.get("order-event.avsc")
    Files.writeString(avscFile, avscText)
    println(
      s"wrote $avscFile: ${Files.readAllLines(avscFile).size} lines, starting with ${Files.readAllLines(avscFile).get(0)}"
    )
  }
}

Encode and Decode One Event

AvroCodec#encode turns a value into a byte array, and AvroCodec#decode turns the bytes back into a value. To encode a placed order:

val event: OrderEvent = OrderPlaced(
  orderId = 42L,
  customer = "Ada",
  amount = BigDecimal("19.99"),
  placedAt = Instant.ofEpochSecond(1700000000L, 123000000),
  note = Some("gift"),
  items = List(LineItem("sku-1", 2, BigDecimal("9.995")))
)

val bytes = codec.encode(event)

To print the bytes in hexadecimal and decode them back:

println(s"${bytes.length} bytes: ${bytes.map(b => f"$b%02x").mkString(" ")}")
// 42 bytes: 00 54 06 41 64 61 04 07 cf 04 44 0c 80 c4 9f d5 0c 80 d3 a6 75 02 08 67 69 66 74 02 0a 73 6b 75 2d 31 04 04 27 0b 06 44 0c 00

println(codec.decode(bytes))
// Right(OrderPlaced(42,Ada,19.99,2023-11-14T22:13:20.123Z,Some(gift),List(LineItem(sku-1,2,9.995))))

The bytes hold values only; the field names live in the schema. The first byte, 00, is the index of the OrderPlaced branch of the union. Then come the order id 54 (42 as a zig-zag variable-length number) and the customer 06 41 64 61 (a length of 3, then the letters Ada). AvroCodec#decode returns an Either[SchemaError, OrderEvent], so a failure is a value to handle, not an exception.

Runnable version, with the codec from “Write the .avsc Schema File”: Step2EncodeDecodeExample.scala

schema-examples/src/main/scala/avrorecords/Step2EncodeDecodeExample.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema-avro:0.0.56
//> using options -experimental
//> using file Model.scala
// Run from this folder:  scala-cli run Step2EncodeDecodeExample.scala
// Or from schema-examples:  sbt "runMain avrorecords.Step2EncodeDecodeExample"

package avrorecords

import zio.blocks.schema.avro._
import java.time.Instant

/**
 * How to Encode and Decode Records with Avro — Encode and Decode One Event
 *
 * Encodes one `OrderPlaced` event, prints the bytes in hexadecimal and decodes
 * them back.
 */
object Step2EncodeDecodeExample {
  def main(args: Array[String]): Unit = {
    // Setup from "Write the .avsc Schema File": the codec
    val codec = OrderEvent.schema.derive(AvroFormat)

    val event: OrderEvent = OrderPlaced(
      orderId = 42L,
      customer = "Ada",
      amount = BigDecimal("19.99"),
      placedAt = Instant.ofEpochSecond(1700000000L, 123000000),
      note = Some("gift"),
      items = List(LineItem("sku-1", 2, BigDecimal("9.995")))
    )

    val bytes = codec.encode(event)

    println(s"${bytes.length} bytes: ${bytes.map(b => f"$b%02x").mkString(" ")}")

    println(codec.decode(bytes))
  }
}

Write Several Events to One Stream

AvroCodec#encode(value, OutputStream) writes one record to any OutputStream, such as a file, a socket or a response body. To write three shipped orders to one buffer:

val shipped: List[OrderEvent] = List(
  OrderShipped(1L, LocalDate.of(2024, 1, 1), "DHL"),
  OrderShipped(2L, LocalDate.of(2024, 1, 2), "UPS"),
  OrderShipped(3L, LocalDate.of(2024, 1, 3), "FedEx")
)

val buffer = new ByteArrayOutputStream()

shipped.foreach(e => codec.encode(e, buffer))

val input = new ByteArrayInputStream(buffer.toByteArray)

AvroCodec#decode(InputStream) reads exactly one record and leaves the rest of the stream unread, so each call continues where the previous one stopped. To read the three events back:

println(s"stream holds ${buffer.size()} bytes")
// stream holds 32 bytes

shipped.foreach(_ => println(codec.decode(input)))
// Right(OrderShipped(1,2024-01-01,DHL))
// Right(OrderShipped(2,2024-01-02,UPS))
// Right(OrderShipped(3,2024-01-03,FedEx))

println(s"bytes left in the stream: ${input.available()}")
// bytes left in the stream: 0

The 32 bytes are the three records back to back, with no length and no separator between them. The stream carries no count and no end marker either, so the reader must know when to stop; here we read as many events as we wrote.

Runnable version, with the codec from “Write the .avsc Schema File”: Step3StreamEventsExample.scala

schema-examples/src/main/scala/avrorecords/Step3StreamEventsExample.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema-avro:0.0.56
//> using options -experimental
//> using file Model.scala
// Run from this folder:  scala-cli run Step3StreamEventsExample.scala
// Or from schema-examples:  sbt "runMain avrorecords.Step3StreamEventsExample"

package avrorecords

import zio.blocks.schema.avro._
import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
import java.time.LocalDate

/**
 * How to Encode and Decode Records with Avro — Write Several Events to One
 * Stream
 *
 * Writes three `OrderShipped` events to one stream, back to back, and reads
 * them back one record at a time.
 */
object Step3StreamEventsExample {
  def main(args: Array[String]): Unit = {
    // Setup from "Write the .avsc Schema File": the codec
    val codec = OrderEvent.schema.derive(AvroFormat)

    val shipped: List[OrderEvent] = List(
      OrderShipped(1L, LocalDate.of(2024, 1, 1), "DHL"),
      OrderShipped(2L, LocalDate.of(2024, 1, 2), "UPS"),
      OrderShipped(3L, LocalDate.of(2024, 1, 3), "FedEx")
    )

    val buffer = new ByteArrayOutputStream()

    shipped.foreach(e => codec.encode(e, buffer))

    val input = new ByteArrayInputStream(buffer.toByteArray)

    println(s"stream holds ${buffer.size()} bytes")

    shipped.foreach(_ => println(codec.decode(input)))

    println(s"bytes left in the stream: ${input.available()}")
  }
}

Handle Bad Input

AvroCodec#decode reports malformed bytes as a SchemaError that names the path of the field it was reading. When the bytes end too early, the error points at the field where the input ran out:

println(codec.decode(bytes.take(10)))
// Left(zio.blocks.schema.SchemaError: Unexpected end of input at: .when[OrderPlaced].amount)

Extra bytes after a complete record are not an error. AvroCodec#decode returns the record and ignores the rest:

println(codec.decode(bytes ++ Array[Byte](1, 2, 3)))
// Right(OrderPlaced(42,Ada,19.99,2023-11-14T22:13:20.123Z,Some(gift),List(LineItem(sku-1,2,9.995))))

warning

Two records in one byte array decode as the first record, with no error. Give every message its own byte array, or frame the messages yourself, for example with a length prefix.

A collection whose length prefix is impossibly large fails before any element is read. The limit is AvroCodec.maxCollectionSize, 2147483639 elements. To build such a prefix with Apache Avro's encoder and decode it as a list:

import org.apache.avro.io.EncoderFactory

val hugeList = new ByteArrayOutputStream()
val prefixEncoder = EncoderFactory.get.binaryEncoder(hugeList, null)
prefixEncoder.writeLong(AvroCodec.maxCollectionSize.toLong + 1L)
prefixEncoder.writeLong(0L)
prefixEncoder.flush()

The decoder reads the length prefix and stops there:

println(Schema[List[Int]].derive(AvroFormat).decode(hugeList.toByteArray))
// Left(zio.blocks.schema.SchemaError: Expected collection size not greater than 2147483639, got 2147483640 at: .)

Runnable version, with the event and bytes from the earlier sections: Step4BadInputExample.scala

schema-examples/src/main/scala/avrorecords/Step4BadInputExample.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema-avro:0.0.56
//> using options -experimental
//> using file Model.scala
// Run from this folder:  scala-cli run Step4BadInputExample.scala
// Or from schema-examples:  sbt "runMain avrorecords.Step4BadInputExample"

package avrorecords

import org.apache.avro.io.EncoderFactory
import zio.blocks.schema._
import zio.blocks.schema.avro._
import java.io.ByteArrayOutputStream
import java.time.Instant

/**
 * How to Encode and Decode Records with Avro — Handle Bad Input
 *
 * Decodes truncated bytes, bytes with extra data at the end, and a list whose
 * length prefix is above `AvroCodec.maxCollectionSize`.
 */
object Step4BadInputExample {
  def main(args: Array[String]): Unit = {
    // Setup from "Write the .avsc Schema File" and "Encode and Decode One Event": the codec and the bytes of one event
    val codec             = OrderEvent.schema.derive(AvroFormat)
    val event: OrderEvent = OrderPlaced(
      orderId = 42L,
      customer = "Ada",
      amount = BigDecimal("19.99"),
      placedAt = Instant.ofEpochSecond(1700000000L, 123000000),
      note = Some("gift"),
      items = List(LineItem("sku-1", 2, BigDecimal("9.995")))
    )
    val bytes = codec.encode(event)

    println(codec.decode(bytes.take(10)))

    println(codec.decode(bytes ++ Array[Byte](1, 2, 3)))

    val hugeList      = new ByteArrayOutputStream()
    val prefixEncoder = EncoderFactory.get.binaryEncoder(hugeList, null)
    prefixEncoder.writeLong(AvroCodec.maxCollectionSize.toLong + 1L)
    prefixEncoder.writeLong(0L)
    prefixEncoder.flush()

    println(Schema[List[Int]].derive(AvroFormat).decode(hugeList.toByteArray))
  }
}

Check How Scala Types Map to Avro

A reader in Java, Python or another Avro library works from the .avsc file, so it sees each Scala type as the Avro type that the codec derives. To print that Avro type for the common Scala types, we define a small helper and a few sample types, a sealed trait of case objects, a one-field case class, and an id built with Schema#transform:

import zio.blocks.chunk.Chunk
import java.util.UUID

def avsc[A](implicit schema: Schema[A]): String = schema.derive(AvroFormat).avroSchema.toString

sealed trait Status
case object Active extends Status
case object Closed extends Status

case class CustomerId(value: Long)

case class Sku(value: String)

implicit val statusSchema: Schema[Status]         = Schema.derived
implicit val customerIdSchema: Schema[CustomerId] = Schema.derived
implicit val skuSchema: Schema[Sku]               = Schema[String].transform(Sku(_), _.value)
implicit val bytesSchema: Schema[Array[Byte]]     = Schema.derived
implicit val chunkSchema: Schema[Chunk[Byte]]     = Schema.derived

To print the Avro type of each one:

List(
  "Int"                 -> avsc[Int],
  "Long"                -> avsc[Long],
  "Short"               -> avsc[Short],
  "Byte"                -> avsc[Byte],
  "Char"                -> avsc[Char],
  "Float"               -> avsc[Float],
  "Double"              -> avsc[Double],
  "Boolean"             -> avsc[Boolean],
  "String"              -> avsc[String],
  "Unit"                -> avsc[Unit],
  "BigInt"              -> avsc[BigInt],
  "BigDecimal"          -> avsc[BigDecimal],
  "Array[Byte]"         -> avsc[Array[Byte]],
  "Chunk[Byte]"         -> avsc[Chunk[Byte]],
  "List[Int]"           -> avsc[List[Int]],
  "Map[String, Int]"    -> avsc[Map[String, Int]],
  "Instant"             -> avsc[Instant],
  "LocalDate"           -> avsc[LocalDate],
  "UUID"                -> avsc[UUID],
  "Option[Int]"         -> avsc[Option[Int]],
  "Either[String, Int]" -> avsc[Either[String, Int]],
  "Status"              -> avsc[Status],
  "CustomerId"          -> avsc[CustomerId],
  "Sku"                 -> avsc[Sku]
).foreach { case (scalaType, avroType) => println(s"$scalaType: $avroType") }
// Int: "int"
// Long: "long"
// Short: "int"
// Byte: "int"
// Char: "int"
// Float: "float"
// Double: "double"
// Boolean: "boolean"
// String: "string"
// Unit: "null"
// BigInt: "bytes"
// BigDecimal: {"type":"record","name":"BigDecimal","namespace":"scala","fields":[{"name":"mantissa","type":"bytes"},{"name":"scale","type":"int"},{"name":"precision","type":"int"},{"name":"roundingMode","type":"int"}]}
// Array[Byte]: {"type":"array","items":"int"}
// Chunk[Byte]: {"type":"array","items":"int"}
// List[Int]: {"type":"array","items":"int"}
// Map[String, Int]: {"type":"map","values":"int"}
// Instant: {"type":"record","name":"Instant","namespace":"java.time","fields":[{"name":"epochSecond","type":"long"},{"name":"nano","type":"int"}]}
// LocalDate: {"type":"record","name":"LocalDate","namespace":"java.time","fields":[{"name":"year","type":"int"},{"name":"month","type":"int"},{"name":"day","type":"int"}]}
// UUID: {"type":"fixed","name":"UUID","namespace":"java.util","size":16}
// Option[Int]: [{"type":"record","name":"None","namespace":"scala","fields":[]},{"type":"record","name":"Some","namespace":"scala","fields":[{"name":"value","type":"int"}]}]
// Either[String, Int]: [{"type":"record","name":"Left","namespace":"scala.util","fields":[{"name":"value","type":"string"}]},{"type":"record","name":"Right","namespace":"scala.util","fields":[{"name":"value","type":"int"}]}]
// Status: [{"type":"record","name":"Active","namespace":"repl.MdocSession.MdocApp","fields":[]},{"type":"record","name":"Closed","namespace":"repl.MdocSession.MdocApp","fields":[]}]
// CustomerId: {"type":"record","name":"CustomerId","namespace":"repl.MdocSession.MdocApp","fields":[{"name":"value","type":"long"}]}
// Sku: "string"

The table sums up the output and what a reader in another language should know about each type:

Scala type	Avro type	What a reader in another language sees
`Int`, `Short`, `Byte`	`int`	A 32-bit number; `Short` and `Byte` are widened
`Long`	`long`
`Float`, `Double`	`float`, `double`
`Boolean`	`boolean`
`String`	`string`
`Char`	`int`	The character's code, not a one-letter string
`Unit`	`null`
`BigInt`	`bytes`	Two's-complement big-endian bytes
`BigDecimal`	record `scala.BigDecimal`	`mantissa` bytes plus `scale`, `precision` and `roundingMode`; not the `decimal` logical type
`Array[Byte]`, `Chunk[Byte]`	array of `int`	One number per byte, not `bytes`
`List[A]`	array of `A`
`Map[String, A]`	map of `A`
`Instant`	record `java.time.Instant`	`epochSecond` and `nano`; not the `timestamp-millis` logical type
`LocalDate`	record `java.time.LocalDate`	`year`, `month` and `day`; not the `date` logical type
`UUID`	`fixed` of 16 bytes named `java.util.UUID`	Not the `uuid` logical type
`Option[A]`	union of records `scala.None` and `scala.Some`	Not `["null", A]`; `None` is branch 0 and `Some` holds `value`
`Either[A, B]`	union of records `scala.util.Left` and `scala.util.Right`	Each side holds `value`
Sealed trait	union of one record per case, in declaration order	A case object is a record with no fields
Case class	record named after the class, namespace from its package	A one-field class such as `CustomerId` stays a record
Newtype built with `Schema#transform`	the wrapped type	`Sku` is a plain `string`

Four rows surprise readers in other languages. Option is a union of two records, not the ["null", T] union that most Avro libraries write for an optional field, so a reader that expects null for a missing value does not find one. Array[Byte] is an array of int rather than bytes, so a byte payload comes out as a list of numbers. BigDecimal, Instant and LocalDate are records and UUID is a 16-byte fixed, not the decimal, timestamp-millis, date and uuid logical types. And Char is a number, not a one-letter string. zio-blocks does not write those other forms, so if another team expects them, agree on the layout with them using the .avsc file.

Runnable version, as one program: Step5TypeMappingExample.scala

schema-examples/src/main/scala/avrorecords/Step5TypeMappingExample.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema-avro:0.0.56
//> using options -experimental
// Run from this folder:  scala-cli run Step5TypeMappingExample.scala
// Or from schema-examples:  sbt "runMain avrorecords.Step5TypeMappingExample"

package avrorecords

import zio.blocks.chunk.Chunk
import zio.blocks.schema._
import zio.blocks.schema.avro._
import java.time.{Instant, LocalDate}
import java.util.UUID

/**
 * How to Encode and Decode Records with Avro — Check How Scala Types Map to
 * Avro
 *
 * Prints the Avro schema that the codec derives for common Scala types.
 */
object Step5TypeMappingExample {

  def avsc[A](implicit schema: Schema[A]): String = schema.derive(AvroFormat).avroSchema.toString

  sealed trait Status
  case object Active extends Status
  case object Closed extends Status

  case class CustomerId(value: Long)

  case class Sku(value: String)

  implicit val statusSchema: Schema[Status]         = Schema.derived
  implicit val customerIdSchema: Schema[CustomerId] = Schema.derived
  implicit val skuSchema: Schema[Sku]               = Schema[String].transform(Sku(_), _.value)
  implicit val bytesSchema: Schema[Array[Byte]]     = Schema.derived
  implicit val chunkSchema: Schema[Chunk[Byte]]     = Schema.derived

  def main(args: Array[String]): Unit =
    List(
      "Int"                 -> avsc[Int],
      "Long"                -> avsc[Long],
      "Short"               -> avsc[Short],
      "Byte"                -> avsc[Byte],
      "Char"                -> avsc[Char],
      "Float"               -> avsc[Float],
      "Double"              -> avsc[Double],
      "Boolean"             -> avsc[Boolean],
      "String"              -> avsc[String],
      "Unit"                -> avsc[Unit],
      "BigInt"              -> avsc[BigInt],
      "BigDecimal"          -> avsc[BigDecimal],
      "Array[Byte]"         -> avsc[Array[Byte]],
      "Chunk[Byte]"         -> avsc[Chunk[Byte]],
      "List[Int]"           -> avsc[List[Int]],
      "Map[String, Int]"    -> avsc[Map[String, Int]],
      "Instant"             -> avsc[Instant],
      "LocalDate"           -> avsc[LocalDate],
      "UUID"                -> avsc[UUID],
      "Option[Int]"         -> avsc[Option[Int]],
      "Either[String, Int]" -> avsc[Either[String, Int]],
      "Status"              -> avsc[Status],
      "CustomerId"          -> avsc[CustomerId],
      "Sku"                 -> avsc[Sku]
    ).foreach { case (scalaType, avroType) => println(s"$scalaType: $avroType") }
}

Read the Bytes with Apache Avro

The proof that the bytes are standard Avro is to read them with another Avro library. Apache Avro's GenericDatumReader reads them with the schema that zio-blocks emits, parsed by Apache Avro's own parser:

import org.apache.avro.Schema.Parser
import org.apache.avro.generic.{GenericData, GenericDatumReader, GenericDatumWriter}
import org.apache.avro.io.DecoderFactory

val avroSchema = new Parser().parse(codec.avroSchema.toString)

val reader = new GenericDatumReader[AnyRef](avroSchema)

def readWithApacheAvro(bytes: Array[Byte]): AnyRef =
  reader.read(null, DecoderFactory.get.binaryDecoder(bytes, null))

To read a shipped order and the placed order from the earlier sections:

println(readWithApacheAvro(codec.encode(shipped.head)))
// {"orderId": 1, "shippedOn": {"year": 2024, "month": 1, "day": 1}, "carrier": "DHL"}

println(readWithApacheAvro(bytes))
// {"orderId": 42, "customer": "Ada", "amount": {"mantissa": "\u0007Ï", "scale": 2, "precision": 34, "roundingMode": 6}, "placedAt": {"epochSecond": 1700000000, "nano": 123000000}, "note": {"value": "gift"}, "items": [{"sku": "sku-1", "quantity": 2, "unitPrice": {"mantissa": "'\u000B", "scale": 3, "precision": 34, "roundingMode": 6}}]}

Apache Avro returns a GenericRecord for each event. The placed order shows how the types from the previous section look to a Java reader: amount is a record of raw mantissa bytes, scale, precision and roundingMode, placedAt is a record of seconds and nanoseconds, and note is the Some record with its value.

Runnable version, with the codec and events from the earlier sections: Step6ReadWithApacheAvroExample.scala

schema-examples/src/main/scala/avrorecords/Step6ReadWithApacheAvroExample.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema-avro:0.0.56
//> using options -experimental
//> using file Model.scala
// Run from this folder:  scala-cli run Step6ReadWithApacheAvroExample.scala
// Or from schema-examples:  sbt "runMain avrorecords.Step6ReadWithApacheAvroExample"

package avrorecords

import org.apache.avro.Schema.Parser
import org.apache.avro.generic.GenericDatumReader
import org.apache.avro.io.DecoderFactory
import zio.blocks.schema.avro._
import java.time.{Instant, LocalDate}

/**
 * How to Encode and Decode Records with Avro — Read the Bytes with Apache Avro
 *
 * Reads bytes written by zio-blocks with Apache Avro's `GenericDatumReader`,
 * using the schema that zio-blocks emits.
 */
object Step6ReadWithApacheAvroExample {
  def main(args: Array[String]): Unit = {
    // Setup from the earlier sections: the codec, a shipped order and the bytes of a placed order
    val codec                     = OrderEvent.schema.derive(AvroFormat)
    val shipped: List[OrderEvent] = List(OrderShipped(1L, LocalDate.of(2024, 1, 1), "DHL"))
    val event: OrderEvent         = OrderPlaced(
      orderId = 42L,
      customer = "Ada",
      amount = BigDecimal("19.99"),
      placedAt = Instant.ofEpochSecond(1700000000L, 123000000),
      note = Some("gift"),
      items = List(LineItem("sku-1", 2, BigDecimal("9.995")))
    )
    val bytes = codec.encode(event)

    val avroSchema = new Parser().parse(codec.avroSchema.toString)

    val reader = new GenericDatumReader[AnyRef](avroSchema)

    def readWithApacheAvro(bytes: Array[Byte]): AnyRef =
      reader.read(null, DecoderFactory.get.binaryDecoder(bytes, null))

    println(readWithApacheAvro(codec.encode(shipped.head)))

    println(readWithApacheAvro(bytes))
  }
}

Write Bytes with Apache Avro

The other direction is to write with Apache Avro and read with zio-blocks. The second branch of the union is the OrderShipped record, so we fill a generic record of that branch and write it with GenericDatumWriter:

val shippedSchema = avroSchema.getTypes.get(1)

val shippedOn = new GenericData.Record(shippedSchema.getField("shippedOn").schema())
shippedOn.put("year", 2024)
shippedOn.put("month", 3)
shippedOn.put("day", 9)

val fromJava = new GenericData.Record(shippedSchema)
fromJava.put("orderId", 7L)
fromJava.put("shippedOn", shippedOn)
fromJava.put("carrier", "DHL")

val javaBytes = new ByteArrayOutputStream()
val javaEncoder = EncoderFactory.get.binaryEncoder(javaBytes, null)
new GenericDatumWriter[AnyRef](avroSchema).write(fromJava, javaEncoder)
javaEncoder.flush()

To decode the bytes that Apache Avro wrote:

println(codec.decode(javaBytes.toByteArray))
// Right(OrderShipped(7,2024-03-09,DHL))

Runnable version, with the codec and the parsed schema from the earlier sections: Step7WriteWithApacheAvroExample.scala

schema-examples/src/main/scala/avrorecords/Step7WriteWithApacheAvroExample.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema-avro:0.0.56
//> using options -experimental
//> using file Model.scala
// Run from this folder:  scala-cli run Step7WriteWithApacheAvroExample.scala
// Or from schema-examples:  sbt "runMain avrorecords.Step7WriteWithApacheAvroExample"

package avrorecords

import org.apache.avro.Schema.Parser
import org.apache.avro.generic.{GenericData, GenericDatumWriter}
import org.apache.avro.io.EncoderFactory
import zio.blocks.schema.avro._
import java.io.ByteArrayOutputStream

/**
 * How to Encode and Decode Records with Avro — Write Bytes with Apache Avro
 *
 * Writes an `OrderShipped` record with Apache Avro's `GenericDatumWriter` and
 * decodes the bytes with zio-blocks.
 */
object Step7WriteWithApacheAvroExample {
  def main(args: Array[String]): Unit = {
    // Setup from the earlier sections: the codec and its schema parsed by Apache Avro
    val codec      = OrderEvent.schema.derive(AvroFormat)
    val avroSchema = new Parser().parse(codec.avroSchema.toString)

    val shippedSchema = avroSchema.getTypes.get(1)

    val shippedOn = new GenericData.Record(shippedSchema.getField("shippedOn").schema())
    shippedOn.put("year", 2024)
    shippedOn.put("month", 3)
    shippedOn.put("day", 9)

    val fromJava = new GenericData.Record(shippedSchema)
    fromJava.put("orderId", 7L)
    fromJava.put("shippedOn", shippedOn)
    fromJava.put("carrier", "DHL")

    val javaBytes   = new ByteArrayOutputStream()
    val javaEncoder = EncoderFactory.get.binaryEncoder(javaBytes, null)
    new GenericDatumWriter[AnyRef](avroSchema).write(fromJava, javaEncoder)
    javaEncoder.flush()

    println(codec.decode(javaBytes.toByteArray))
  }
}

What Avro Support Does Not Cover

The module encodes and decodes one value with one schema, and several things that Avro users expect from other libraries are left to you. The codec writes the bytes of a single value. An Avro container file, the .avro file that Apache Avro's DataFileWriter writes with its schema embedded, is not supported, so a reader of a container file needs Apache Avro itself. There is also no registry framing: a schema registry stores your .avsc files under version ids and expects each message to start with the id of its schema, which you add yourself. Build a Schema Registry shows how to store schemas as data in ZIO Blocks. The two cases below need more care, because the codec does not report them.

Old and New Versions of a Type

There is no reader and writer schema resolution: the codec reads bytes with its own schema only, and a default value in the new type is not used to fill a field that the old bytes lack. To see both directions, we define two versions of an order, the second with a new field that has a default:

case class OrderV1(orderId: Long, customer: String)

object OrderV1 {
  implicit val schema: Schema[OrderV1] = Schema.derived
}

case class OrderV2(orderId: Long, customer: String, priority: Int = 0)

object OrderV2 {
  implicit val schema: Schema[OrderV2] = Schema.derived
}

val v1Bytes = OrderV1.schema.derive(AvroFormat).encode(OrderV1(9L, "Bob"))
val v2Bytes = OrderV2.schema.derive(AvroFormat).encode(OrderV2(9L, "Bob", 3))

To read old bytes with the new type, and new bytes with the old type:

println(OrderV2.schema.derive(AvroFormat).decode(v1Bytes))
// Left(zio.blocks.schema.SchemaError: Unexpected end of input at: .priority)

println(OrderV1.schema.derive(AvroFormat).decode(v2Bytes))
// Right(OrderV1(9,Bob))

Old bytes fail with an error at the new field, and new bytes read by the old type lose the new field without an error. To change a type safely, convert between versions in Scala with Into, As or Migration, as shown in Evolve Your Data.

Bytes Written by a Different Type

Avro bytes carry no schema id and no fingerprint, so nothing stops a reader from decoding bytes with the wrong schema. To decode the bytes of a shipped order as an OrderV1:

println(OrderV1.schema.derive(AvroFormat).decode(codec.encode(shipped.head)))
// Right(OrderV1(1,�))

The decode succeeds and returns nonsense. The union index becomes the orderId, the real order id becomes the length of customer, and customer holds one byte of the date; the remaining bytes are ignored, as in the previous section. Keep track of which schema wrote the bytes, for example with one topic or file per event type, a version field, or a schema registry.

Runnable version, with the codec and events from the earlier sections: Step8AvroLimitsExample.scala

schema-examples/src/main/scala/avrorecords/Step8AvroLimitsExample.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema-avro:0.0.56
//> using options -experimental
//> using file Model.scala
// Run from this folder:  scala-cli run Step8AvroLimitsExample.scala
// Or from schema-examples:  sbt "runMain avrorecords.Step8AvroLimitsExample"

package avrorecords

import zio.blocks.schema._
import zio.blocks.schema.avro._
import java.time.LocalDate

/**
 * How to Encode and Decode Records with Avro — What Avro Support Does Not Cover
 *
 * Reads old bytes with a new version of a type and new bytes with the old
 * version, then decodes the bytes of one type as another type.
 */
object Step8AvroLimitsExample {

  case class OrderV1(orderId: Long, customer: String)

  object OrderV1 {
    implicit val schema: Schema[OrderV1] = Schema.derived
  }

  case class OrderV2(orderId: Long, customer: String, priority: Int = 0)

  object OrderV2 {
    implicit val schema: Schema[OrderV2] = Schema.derived
  }

  def main(args: Array[String]): Unit = {
    // Setup from the earlier sections: the codec and a shipped order
    val codec                     = OrderEvent.schema.derive(AvroFormat)
    val shipped: List[OrderEvent] = List(OrderShipped(1L, LocalDate.of(2024, 1, 1), "DHL"))

    val v1Bytes = OrderV1.schema.derive(AvroFormat).encode(OrderV1(9L, "Bob"))
    val v2Bytes = OrderV2.schema.derive(AvroFormat).encode(OrderV2(9L, "Bob", 3))

    println(OrderV2.schema.derive(AvroFormat).decode(v1Bytes))

    println(OrderV1.schema.derive(AvroFormat).decode(v2Bytes))

    println(OrderV1.schema.derive(AvroFormat).decode(codec.encode(shipped.head)))
  }
}

Putting It Together

The complete program derives the codec, prints the .avsc schema of the whole union, writes a stream of three events to a buffer, reads them back, and checks the round trip. It differs from the sections in three ways: the stream mixes an OrderPlaced, an OrderShipped and an OrderCancelled event, where the stream section wrote only shipped orders; it prints the schema of the whole union rather than of OrderCancelled; and it ends by comparing the decoded events with the originals:

schema-examples/src/main/scala/avrorecords/CompleteExample.scala
//> using scala 3.9.0
//> using dep dev.zio::zio-blocks-schema-avro:0.0.56
//> using options -experimental
//> using file Model.scala
// Run from this folder:  scala-cli run CompleteExample.scala
// Or from schema-examples:  sbt "runMain avrorecords.CompleteExample"

package avrorecords

import org.apache.avro.SchemaFormatter
import zio.blocks.schema.avro._
import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
import java.time.{Instant, LocalDate}

/**
 * How to Encode and Decode Records with Avro — Putting It Together
 *
 * Derives the codec, prints the `.avsc` schema of the whole sealed trait,
 * writes three different events to one stream and reads them back.
 */
object CompleteExample {
  def main(args: Array[String]): Unit = {
    val codec = OrderEvent.schema.derive(AvroFormat)

    println(SchemaFormatter.format("json/pretty", codec.avroSchema))

    val events: List[OrderEvent] = List(
      OrderPlaced(
        orderId = 42L,
        customer = "Ada",
        amount = BigDecimal("19.99"),
        placedAt = Instant.ofEpochSecond(1700000000L, 123000000),
        note = Some("gift"),
        items = List(LineItem("sku-1", 2, BigDecimal("9.995")))
      ),
      OrderShipped(42L, LocalDate.of(2024, 1, 3), "DHL"),
      OrderCancelled(43L, "customer request")
    )

    val buffer = new ByteArrayOutputStream()

    events.foreach(e => codec.encode(e, buffer))

    val input = new ByteArrayInputStream(buffer.toByteArray)

    println(s"stream holds ${buffer.size()} bytes")

    val decoded = events.map(_ => codec.decode(input))

    decoded.foreach(println)

    println(s"decoded events equal the originals: ${decoded == events.map(Right(_))}")
  }
}

Running it prints the following; the .avsc text, 120 lines in full, is cut after its first field, where the ... line stands:

[ {
  "type" : "record",
  "name" : "OrderPlaced",
  "namespace" : "avrorecords",
  "fields" : [ {
    "name" : "orderId",
    "type" : "long"
...
stream holds 71 bytes
Right(OrderPlaced(42,Ada,19.99,2023-11-14T22:13:20.123Z,Some(gift),List(LineItem(sku-1,2,9.995))))
Right(OrderShipped(42,2024-01-03,DHL))
Right(OrderCancelled(43,customer request))
decoded events equal the originals: true

Running the Examples

All runnable code from this guide is in schema-examples/src/main/scala/avrorecords/. Run any file from that folder with scala-cli run <File>.scala, or from schema-examples with sbt "runMain avrorecords.<ObjectName>".

Going Further

The Avro codec reference covers every method of AvroCodec, the error model and recursive types.
One Type, Many Formats derives JSON, MessagePack, Avro and other codecs from one schema and compares their sizes.
Build a Schema Registry stores versioned event schemas as data, the usual next step once several services exchange Avro messages.
Evolve Your Data changes a type between versions with Into, As and Migration.
The AvroListOfRecordsBenchmark in the repository's benchmarks module compares this codec with avro4s and zio-schema-avro.

The Problem​

Prerequisites​

The Order Events​

Write the .avsc Schema File​

Encode and Decode One Event​

Write Several Events to One Stream​

Handle Bad Input​

Check How Scala Types Map to Avro​

Read the Bytes with Apache Avro​

Write Bytes with Apache Avro​

What Avro Support Does Not Cover​

Old and New Versions of a Type​

Bytes Written by a Different Type​

Putting It Together​

Running the Examples​

Going Further​