How to read and write Avro files with Spark

Avro datasets carry a schema with compact binary records, which makes them common at the boundary between Apache Spark jobs, ingestion systems, and message pipelines. Reading and writing them in Spark uses the spark-avro data source module plus the DataFrame reader and writer APIs.

The spark-avro module is built by the Spark project but is still added as an external package for submitted jobs and interactive shells. Once that package is present, PySpark uses format(“avro”) for both reads and writes instead of a separate avro() reader or writer method.

A compact smoke test proves both Spark behavior and the filesystem result by writing Avro, reading the same directory back for schema and row count, writing a filtered output directory, and reading that output for expected values. Treat schema compatibility as a separate contract review when other producers or consumers share the same Avro subject.

1. Create a PySpark script that writes and reads Avro data.

   spark_avro_check.py

   import os
   import shutil
    
   from pyspark.sql import SparkSession
   from pyspark.sql import functions as F
    
   base_path = "/tmp/sg-spark-avro"
   input_path = f"{base_path}/orders"
   output_path = f"{base_path}/priority-apac"
    
   shutil.rmtree(base_path, ignore_errors=True)
    
   spark = (
       SparkSession.builder.master("local[1]")
       .appName("sg-spark-avro-check")
       .getOrCreate()
   )
   spark.sparkContext.setLogLevel("ERROR")
    
   orders = spark.createDataFrame(
       [
           ("ORD-1001", "APAC", 3, True),
           ("ORD-1002", "EMEA", 1, False),
           ("ORD-1003", "APAC", 7, True),
       ],
       "order_id string, region string, item_count int, priority boolean",
   )
    
   orders.write.format("avro").mode("overwrite").save(input_path)
    
   loaded = spark.read.format("avro").load(input_path)
    
   print("Input Avro schema:")
   loaded.printSchema()
   print(f"Input rows: {loaded.count()}")
    
   (
       loaded.filter((F.col("region") == "APAC") & F.col("priority"))
       .select("order_id", "region", "item_count")
       .write.format("avro")
       .mode("overwrite")
       .save(output_path)
   )
    
   result = spark.read.format("avro").load(output_path)
    
   print("Output Avro rows:")
   result.orderBy("order_id").show(truncate=False)
   print("Output files:")
   for name in sorted(os.listdir(output_path)):
       if not name.startswith("."):
           print(name)
    
   spark.stop()

   Spark writes Avro output as a directory that contains one or more part-*.avro data files plus commit metadata such as _SUCCESS.

2. Run the script with the Spark Avro package.

   $ spark-submit \
     --packages org.apache.spark:spark-avro_2.13:4.1.2 \
     spark_avro_check.py
   ##### snipped #####
   Input Avro schema:
   root
    |-- order_id: string (nullable = true)
    |-- region: string (nullable = true)
    |-- item_count: integer (nullable = true)
    |-- priority: boolean (nullable = true)
   
   Input rows: 3
   Output Avro rows:
   +--------+------+----------+
   |order_id|region|item_count|
   +--------+------+----------+
   |ORD-1001|APAC  |3         |
   |ORD-1003|APAC  |7         |
   +--------+------+----------+
   
   Output files:
   _SUCCESS
   part-00000-6f19d40d-25fa-4a1f-ad75-8c1a5e0633da-c000.snappy.avro

   Match the package version to the Spark runtime. A Spark 4.1.2 job uses the spark-avro_2.13:4.1.2 coordinate shown here.
   Related: How to add packages to a Spark job

3. Remove the sample Avro directory after confirming the read-back output.

   $ rm -r /tmp/sg-spark-avro