JSON to Avro Schema Generator

JSON to Avro Schema: Union Types, Logical Types, and Kafka Schema Registry

Apache Avro stores data as compact binary without repeating field names — a 100-byte JSON event typically serializes to 20-30 bytes in Avro, reducing Kafka storage and network costs by 60-80%. The schema (AVSC) is registered separately in Confluent Schema Registry; each Kafka message carries a 5-byte header (magic byte + schema ID) instead of the full schema, and consumers fetch the schema by ID to deserialize. The design decisions that matter most are handling nullable fields correctly (union types with a null default), using logical types for timestamps and decimals (not raw long or double), and understanding schema evolution compatibility rules before adding or removing fields in production.

Live Example: Complete Avro Schema

{
  "type":      "record",
  "name":      "OrderEvent",
  "namespace": "com.example.orders.v1",
  "doc":       "Event fired when an order is created or updated",
  "fields": [
    {
      "name": "order_id",
      "type": "string",
      "doc":  "UUID primary key"
    },
    {
      "name": "user_id",
      "type": "string"
    },
    {
      "name":    "status",
      "type":    {
        "type":    "enum",
        "name":    "OrderStatus",
        "symbols": ["PENDING", "CONFIRMED", "SHIPPED", "DELIVERED", "CANCELLED"]
      },
      "default": "PENDING"
    },
    {
      "name": "amount",
      "type": {
        "type":        "bytes",
        "logicalType": "decimal",
        "precision":   10,
        "scale":       2
      },
      "doc": "Order total — use decimal, NOT double, for money"
    },
    {
      "name":    "discount_code",
      "type":    ["null", "string"],
      "default": null,
      "doc":     "Optional — union with null first, default null"
    },
    {
      "name": "items",
      "type": {
        "type":  "array",
        "items": {
          "type":   "record",
          "name":   "OrderItem",
          "fields": [
            { "name": "sku",      "type": "string" },
            { "name": "quantity", "type": "int" },
            { "name": "price",    "type": { "type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2 } }
          ]
        }
      },
      "default": []
    },
    {
      "name": "metadata",
      "type": { "type": "map", "values": "string" },
      "default": {},
      "doc": "Arbitrary string key-value pairs (device, utm_source, etc.)"
    },
    {
      "name": "created_at",
      "type": {
        "type":        "long",
        "logicalType": "timestamp-millis"
      },
      "doc": "Milliseconds since Unix epoch"
    },
    {
      "name":    "updated_at",
      "type":    ["null", { "type": "long", "logicalType": "timestamp-millis" }],
      "default": null
    }
  ]
}

Avro Type System Reference

{
  // Primitive types
  "null":    null,       // absence of value — must be first in a union for null default
  "boolean": true,
  "int":     32,         // 32-bit signed integer
  "long":    1000000,    // 64-bit signed integer — use for IDs, timestamps
  "float":   1.5,        // 32-bit IEEE 754 — imprecise, avoid for money
  "double":  3.14,       // 64-bit IEEE 754 — imprecise, avoid for money
  "bytes":   "...",      // raw binary (also used with decimal logicalType)
  "string":  "hello",

  // Complex types
  "record": { "type": "record", "name": "MyRecord", "fields": [...] },
  "enum":   { "type": "enum", "name": "Status", "symbols": ["A", "B", "C"] },
  "array":  { "type": "array", "items": "string" },
  "map":    { "type": "map",   "values": "string" },  // keys are always strings
  "fixed":  { "type": "fixed", "size": 16, "name": "MD5" },  // fixed-size bytes

  // Union — JSON null maps to ["null", "actualType"]
  "union":  ["null", "string"],   // nullable string
  // RULE: null must be FIRST in the union for default: null to be valid

  // Logical types — overlay semantics on primitive types
  "decimal":          { "type": "bytes",  "logicalType": "decimal",          "precision": 10, "scale": 2 },
  "date":             { "type": "int",    "logicalType": "date" },            // days since 1970-01-01
  "time-millis":      { "type": "int",    "logicalType": "time-millis" },     // ms since midnight
  "timestamp-millis": { "type": "long",   "logicalType": "timestamp-millis"}, // ms since epoch (UTC)
  "timestamp-micros": { "type": "long",   "logicalType": "timestamp-micros"}, // μs since epoch
  "uuid":             { "type": "string", "logicalType": "uuid" }             // UUID string
}

Nullable Fields: The Union Pattern

{
  "fields": [
    // REQUIRED field — no union, no default
    { "name": "order_id", "type": "string" },

    // OPTIONAL nullable field — union, default MUST be null (first type in union)
    { "name": "discount_code", "type": ["null", "string"], "default": null },

    // OPTIONAL with fallback value
    { "name": "currency", "type": ["string", "null"], "default": "USD" },
    // NOTE: here "string" is first → default "USD" is valid
    // If you want default null, "null" MUST be first

    // OPTIONAL nullable record
    {
      "name":    "shipping_address",
      "type":    ["null", {
        "type":   "record",
        "name":   "Address",
        "fields": [
          { "name": "street",  "type": "string" },
          { "name": "city",    "type": "string" },
          { "name": "country", "type": "string" }
        ]
      }],
      "default": null
    }
  ]
}

Schema Evolution: Backward and Forward Compatibility

{
  // BACKWARD COMPATIBLE changes (new schema reads old data):
  // ✅ Add field with default value
  { "name": "campaign_id", "type": ["null", "string"], "default": null },
  // Old records without this field get the default value when read by new schema

  // ✅ Remove field with default value
  // Old records had "deprecated_field" — new schema just ignores it

  // ✅ Change type from int to long (widening)
  // Old: { "name": "count", "type": "int" }
  // New: { "name": "count", "type": "long" }   // safe widening

  // BREAKING changes (avoid in production):
  // ❌ Add field WITHOUT a default
  { "name": "required_new_field", "type": "string" },
  // Old records missing this field fail to deserialize

  // ❌ Rename a field (not the same as adding aliases)
  // Use aliases instead:
  { "name": "user_identifier", "aliases": ["user_id"], "type": "string" },
  // Old records with user_id are mapped to user_identifier

  // ❌ Change type incompatibly (string → int)
  // ❌ Remove field without alias (old code expecting it will fail)
  // ❌ Change enum symbols (reorder or remove)
}

// Confluent Schema Registry compatibility levels:
// BACKWARD     — new schema can read data written with old schema (safest, default)
// FORWARD      — old schema can read data written with new schema
// FULL         — both backward AND forward compatible
// NONE         — no compatibility checks

Kafka Schema Registry Integration

// Register a schema (REST API)
curl -X POST http://localhost:8081/subjects/orders-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"OrderEvent\",...}"}'
// → {"id": 42}

// Producer (Python with confluent-kafka)
from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import SerializationContext, MessageField

schema_registry = SchemaRegistryClient({"url": "http://localhost:8081"})
avro_serializer = AvroSerializer(schema_registry, schema_str)

producer = Producer({"bootstrap.servers": "localhost:9092"})
producer.produce(
    topic="orders",
    key=order_id,
    value=avro_serializer(order_dict, SerializationContext("orders", MessageField.VALUE))
)

// Consumer (Python)
from confluent_kafka.schema_registry.avro import AvroDeserializer

avro_deserializer = AvroDeserializer(schema_registry)
consumer.subscribe(["orders"])
msg = consumer.poll(timeout=1.0)
order = avro_deserializer(msg.value(), SerializationContext("orders", MessageField.VALUE))
print(order["order_id"])   # fully deserialized dict matching the AVSC

Code Generation from AVSC

// Generate Java classes from AVSC (avro-tools)
java -jar avro-tools-1.11.0.jar compile schema order.avsc ./src

// Generated Java:
// public class OrderEvent extends SpecificRecordBase {
//   private CharSequence order_id;
//   private OrderStatus status;
//   ...
// }

// Python: read/write Avro with fastavro
import fastavro
from io import BytesIO

# Write
schema = fastavro.parse_schema(json.load(open("order.avsc")))
records = [{"order_id": "ord_001", "status": "PENDING", ...}]
buf = BytesIO()
fastavro.writer(buf, schema, records)

# Read
buf.seek(0)
for record in fastavro.reader(buf):
    print(record["order_id"])   # typed dict matching schema

# Node.js: avsc library
const avsc = require('avsc');
const OrderSchema = avsc.Type.forSchema(require('./order.avsc'));
const buf = OrderSchema.toBuffer({ orderId: 'ord_001', status: 'PENDING', ... });
const order = OrderSchema.fromBuffer(buf);

Best Practices for Production

• Always put null first in union types and set "default": null: The Avro spec requires the default value to match the first type in a union. If null is first, the default can be null. If the primary type is first, the default must be a value of that type. Failing to follow this rule causes schema registration errors.
• Use logical types for timestamps and decimals: Raw long fields are ambiguous (milliseconds? seconds? microseconds?). logicalType: "timestamp-millis" carries the semantic meaning and lets code generators produce proper datetime and Decimal types. Raw double for money causes rounding errors — use logicalType: "decimal" with explicit precision and scale.
• Set compatibility to BACKWARD in the Schema Registry: BACKWARD compatibility means new schema versions can read data serialized with old schemas. This allows producers to be updated before consumers without dropping messages. BACKWARD is the safe default for most Kafka pipelines.
• Name records with fully qualified names: Use "namespace": "com.example.orders.v1" and avoid name collisions. When a record type is reused in multiple schemas, the namespace + name combination is the globally unique identifier used by the Schema Registry and code generators.

FAQ

Q: Avro vs Protobuf — when to choose each?
A: Avro is the dominant choice in the Kafka ecosystem (Confluent Schema Registry has first-class Avro support). Schema evolution via defaults is ergonomic. Protobuf generates code for 20+ languages and is better for gRPC microservices. Avro schemas are JSON themselves (easy to generate programmatically); Protobuf uses a separate IDL. For Kafka data pipelines, use Avro. For gRPC/HTTP APIs, use Protobuf.

Q: Can I convert Avro data to Parquet?
A: Yes — Avro and Parquet share a similar type system, and the conversion is well-supported. Spark, Hive, and AWS Glue all support reading Avro from Kafka/S3 and writing Parquet to a data lake. The schema translation is straightforward, with Avro records mapping to Parquet groups.

Q: How do I handle Avro schemas in a monorepo?
A: Store .avsc files in a shared schema directory. Run code generation as part of the build (Maven/Gradle plugin for Java, avro-gen for Python). Register schemas to the Schema Registry as part of CI — fail the build if a schema change is backward-incompatible without a version bump.