Skip to main content
Markdown

Migrate a Billing Event Schema

Read historical billing events through a new Avro contract using reader aliases and defaults. The producer emits Payment, while analytics expects BillingEvent, a renamed account field, and a currency. Schema resolution applies those changes during decoding, with Parquet output written in batches.

Uses: Data SDK, Avro resolution, Parquet. Services: none. Difficulty: intermediate.

Read the Old Contract Through the New One​

This fixture's amounts are all USD cents. Adding a currency default is valid only because that is already a known property of the source contract.

migrate.codon
import exaloop.data as data
from exaloop.data.avro import Schema, iter_batches

writer = Schema('''{
  "type":"record", "name":"Payment", "namespace":"billing",
  "fields":[
    {"name":"id", "type":"long"},
    {"name":"account", "type":"string"},
    {"name":"amount_cents", "type":"long"}
  ]
}''')
reader = Schema('''{
  "type":"record", "name":"BillingEvent", "namespace":"billing",
  "aliases":["Payment"],
  "fields":[
    {"name":"id", "type":"long"},
    {"name":"customer_id", "aliases":["account"], "type":"string"},
    {"name":"amount_cents", "type":"long"},
    {"name":"currency", "type":"string", "default":"USD"}
  ]
}''')

with data.table({"id": data.array([101, 102, 103]),
                 "account": data.array(["A", "B", "A"]),
                 "amount_cents": data.array([12500, 7800, 4500])}) as source:
    source.write_avro("payments-v1.avro", writer, codec="zstandard", batch_size=2)

row_count = 0
amount_cents = 0
parts: list[str] = []
for batch in iter_batches("payments-v1.avro", reader_schema=reader,
                          batch_size=2, max_output_bytes=1048576,
                          max_block_bytes=1048576):
    with batch:
        with batch["currency"] as currencies, batch["amount_cents"] as amounts:
            assert all(value == "USD" for value in currencies.to_list(str))
            values = amounts.to_list(int)
            row_count += len(values)
            amount_cents += sum(values)
        filename = "billing-part-" + str(len(parts)) + ".parquet"
        batch.write(filename)
        parts.append(filename)

assert row_count == 3 and amount_cents == 24800 and len(parts) == 2
with data.read_parquet(parts[0]) as first:
    with first["customer_id"] as customers:
        assert customers.to_list(str) == ["A", "B"]
print("rows", row_count, "amount_cents", amount_cents, "parts", len(parts))
Terminal
exaloop --with data run -release migrate.codon
Output
rows 3 amount_cents 24800 parts 2

The two Parquet files contain the reader's field names and currency values. Avro resolution happens natively, and only one output batch is retained at a time. Batch and block limits are separate: the reader must still decompress an entire input block. The byte budgets are not a cap on total process memory.

Turn It Into a Backfill​

Use an immutable input snapshot and a new output directory per run. Record input checksums, reader schema version, part names, row counts, and control totals in a run manifest. Publish that manifest only after every input and output check passes. A failed run may leave valid but incomplete part files; downstream jobs must not discover them by globbing an unfinished directory.

Keep exact amounts and currency together. Defaults fill missing fields; they do not convert currencies, repair bad values, or override an existing field. Add an explicit business validation step for those cases. Record aliases express a deliberate rename; do not rely on namespace changes to distinguish incompatible business concepts with the same short type name.

For a narrow analytical extraction, columns=["id", "amount_cents"] avoids constructing unrelated fields. Projection is not full-file validation because unselected values may be skipped. This recipe intentionally reads every field.