Definition
- Schema evolution – managing changes to data structure over time (adding, removing, or changing fields) without breaking existing producers or consumers.
- Schema Registry – a central service that stores every version of each schema and enforces compatibility rules at registration time.
- Backward compatibility – a new schema can read data written with the old schema (new consumer, old data).
- Forward compatibility – an old schema can read data written with the new schema (old consumer, new data).
- Column pruning – reading only the columns a query needs from a columnar file, skipping the rest entirely.
- Predicate pushdown – using per-block min/max statistics to skip blocks that cannot match a filter.
The National Bank of India keeps changing its transaction slip over the years: one year they add a "channel" field for mobile versus branch, another year they drop an unused box. Schema evolution is making sure an old reader can still process a new slip and a new reader can still process an old slip, so nothing breaks when the form changes. The Schema Registry is the central rule book holding every version of the slip and refusing a change that would break existing staff. And storing slips column by column, like Parquet, is like filing all the amounts in one slim folder, so totalling amounts means grabbing one folder instead of flipping through every full slip.
Understanding the Concepts
What is schema evolution?
Schema evolution is managing changes to data structure over time without breaking existing producers or consumers. Producers and consumers deploy independently, so a producer might start writing a new field before all consumers understand it, or an old consumer might read new data. Without discipline, this breaks pipelines.
package com.bank.schema;
public class SchemaEvolutionOverview {
// Version 1: original transaction event
record TxnV1(String txnId, String accountId,
java.math.BigDecimal amount) {}
// Version 2: added 'channel' field
record TxnV2(String txnId, String accountId,
java.math.BigDecimal amount,
String channel) {}
public static void main(String[] args) {
// V1 data still readable by V2 consumer
// (channel defaults to null)
TxnV1 old = new TxnV1("T1", "SAV2001",
new java.math.BigDecimal("15000"));
System.out.println("V1 event: " + old);
TxnV2 newer = new TxnV2("T2", "SAV2001",
new java.math.BigDecimal("25000"), "MOBILE");
System.out.println("V2 event: " + newer);
}
}
V1 event: TxnV1[txnId=T1, accountId=SAV2001, amount=15000]
V2 event: TxnV2[txnId=T2, accountId=SAV2001, amount=25000, channel=MOBILE]
How does a Schema Registry work?
A Schema Registry is the central service that stores every version of each schema and enforces compatibility rules when a producer tries to register a new version. The registry rejects changes that would break consumers.
package com.bank.schema;
public class RegistryRules {
public static void main(String[] args) {
System.out.println("Compatibility modes:");
System.out.println(" BACKWARD: new schema reads old data");
System.out.println(" -> add fields WITH defaults");
System.out.println(" FORWARD: old schema reads new data");
System.out.println(" -> remove optional fields only");
System.out.println(" FULL: both directions");
System.out.println(" -> add/remove with defaults only");
System.out.println();
System.out.println("Registry ACCEPTS:");
System.out.println(" Add nullable 'channel' with default");
System.out.println("Registry REJECTS:");
System.out.println(" Add mandatory 'channel' (no default)");
System.out.println(" Rename 'amount' to 'txn_amount'");
System.out.println(" Change 'amount' from decimal to string");
}
}
Compatibility modes:
BACKWARD: new schema reads old data
-> add fields WITH defaults
FORWARD: old schema reads new data
-> remove optional fields only
FULL: both directions
-> add/remove with defaults only
Registry ACCEPTS:
Add nullable 'channel' with default
Registry REJECTS:
Add mandatory 'channel' (no default)
Rename 'amount' to 'txn_amount'
Change 'amount' from decimal to string
How do you wire schema evolution into a Kafka pipeline?
Producers serialize with the registry: the serializer fetches or registers the schema id, and each message carries only the schema id rather than the full schema. Consumers fetch the matching schema to deserialize. This keeps messages small and guarantees both sides agree.
package com.bank.schema;
import java.util.Properties;
public class KafkaSchemaWiring {
public static void main(String[] args) {
Properties producerProps = new Properties();
producerProps.put("schema.registry.url",
"http://registry:8081");
producerProps.put("key.serializer",
"io.confluent.kafka.serializers"
+ ".KafkaAvroSerializer");
producerProps.put("value.serializer",
"io.confluent.kafka.serializers"
+ ".KafkaAvroSerializer");
System.out.println("Producer serializes with registry:");
System.out.println(" Message = [schema_id (4 bytes)]"
+ " + [Avro payload]");
System.out.println(" Schema stored centrally, not per msg");
System.out.println();
System.out.println("Consumer deserializes:");
System.out.println(" Reads schema_id -> fetches schema"
+ " from registry -> deserializes payload");
}
}
Producer serializes with registry:
Message = [schema_id (4 bytes)] + [Avro payload]
Schema stored centrally, not per msg
Consumer deserializes:
Reads schema_id -> fetches schema from registry -> deserializes payload
Why do columnar formats matter for analytics?
Columnar formats like Parquet store values column by column instead of row by row. This layout enables three major performance wins for analytical queries.
package com.bank.schema;
public class ColumnarBenefits {
public static void main(String[] args) {
System.out.println("1. Column pruning:");
System.out.println(" Query needs only 'amount' from"
+ " 100-column table.");
System.out.println(" Parquet reads only amount column.");
System.out.println(" Row-store reads all 100 columns.");
System.out.println();
System.out.println("2. Predicate pushdown:");
System.out.println(" Filter: amount > 100000");
System.out.println(" Row group max = 50000 -> SKIP");
System.out.println(" Row group max = 200000 -> READ");
System.out.println();
System.out.println("3. Compression:");
System.out.println(" Same-type values compress 5-10x.");
System.out.println(" Dictionary encoding for branches.");
System.out.println(" Run-length encoding for dates.");
}
}
1. Column pruning:
Query needs only 'amount' from 100-column table.
Parquet reads only amount column.
Row-store reads all 100 columns.
2. Predicate pushdown:
Filter: amount > 100000
Row group max = 50000 -> SKIP
Row group max = 200000 -> READ
3. Compression:
Same-type values compress 5-10x.
Dictionary encoding for branches.
Run-length encoding for dates.
How does column pruning work?
Column pruning means reading only the columns a query needs. On a 100-column transaction table, if the query only selects amount and date, Parquet reads just those two columns' bytes and never touches the other 98, cutting I/O by orders of magnitude.
# Column pruning in action
# Table has 100 columns, query needs only 2
df = spark.read.parquet(
"s3://bank-lake/curated/transactions/")
# This reads ONLY amount and txn_date columns
result = df.select("amount", "txn_date") \
.groupBy("txn_date") \
.sum("amount")
result.explain(True) # Physical plan shows column pruning
== Physical Plan ==
*(1) HashAggregate(keys=[txn_date], functions=[sum(amount)])
+- FileScan parquet [amount, txn_date] <-- only 2 of 100 columns read
PushedFilters: []
ReadSchema: struct<amount:decimal(15,2),txn_date:string>
How does predicate pushdown work?
Predicate pushdown uses per-row-group min/max statistics to skip irrelevant data. When Spark pushes a filter like amount > 100000, the reader checks each row group's max for amount; if the max is below the threshold, the entire row group is skipped without decompression.
# Predicate pushdown with Parquet statistics
df = spark.read.parquet(
"s3://bank-lake/curated/transactions/")
# Filter pushed to Parquet reader
high_value = df.filter("amount > 100000")
high_value.explain(True)
== Physical Plan ==
*(1) Filter (amount > 100000)
+- FileScan parquet [txn_id, amount, ...]
PushedFilters: [GreaterThan(amount, 100000)]
ReadSchema: struct<txn_id:string, amount:decimal(15,2)>
-- Row groups with max(amount) < 100000 are SKIPPED
Why does sort order matter for pushdown?
Min/max statistics only help if the per-block ranges are tight. If data is sorted by amount, block zero holds 0-10000, block one holds 10000-20000, so a filter cleanly isolates blocks. If unsorted, every block holds a random spread and no block is ever skipped.
package com.bank.schema;
public class SortOrderMatters {
public static void main(String[] args) {
System.out.println("Sorted by amount:");
System.out.println(" Block 0: min=0, max=10000");
System.out.println(" Block 1: min=10001, max=20000");
System.out.println(" Block 2: min=20001, max=50000");
System.out.println(" Filter: amount > 30000");
System.out.println(" -> Blocks 0,1 SKIPPED. Block 2 READ.");
System.out.println();
System.out.println("Unsorted:");
System.out.println(" Block 0: min=50, max=49000");
System.out.println(" Block 1: min=100, max=48000");
System.out.println(" Block 2: min=75, max=50000");
System.out.println(" Filter: amount > 30000");
System.out.println(" -> ALL blocks must be read."
+ " Statistics useless.");
}
}
Sorted by amount:
Block 0: min=0, max=10000
Block 1: min=10001, max=20000
Block 2: min=20001, max=50000
Filter: amount > 30000
-> Blocks 0,1 SKIPPED. Block 2 READ.
Unsorted:
Block 0: min=50, max=49000
Block 1: min=100, max=48000
Block 2: min=75, max=50000
Filter: amount > 30000
-> ALL blocks must be read. Statistics useless.
What are common interview follow-up questions?
- What is the difference between backward and forward compatibility?
- How does Avro handle schema evolution differently from Parquet?
- What is Z-ordering and how does it help multi-column pushdown?
- How do lakehouse formats (Delta, Iceberg) handle schema evolution on existing data?
- What is dictionary encoding and when does it help?
Visual Overview
Schema Evolution in a Kafka Pipeline
────────────────────────────────────
Producer Registry Consumer
┌──────────┐ ┌──────────────┐ ┌──────────┐
│ Serialize│────────▶│ Store schema │◀───────│Deserialize│
│ with │ register│ Version 1 │ fetch │ with │
│ schema v2│ │ Version 2 │ │ schema v2│
└──────────┘ │ (compat │ └──────────┘
│ │ rules) │ │
▼ └──────────────┘ ▼
[schema_id + payload] [payload -> record]
Columnar Format Benefits (Parquet)
──────────────────────────────────
Row-Store (CSV/OLTP) Column-Store (Parquet)
┌───────────────────┐ ┌──────────────────────┐
│ id|amt|dt|branch │ │ id: 1, 2, 3, ... │
│ id|amt|dt|branch │ │ amt: 100, 200, ... │
│ id|amt|dt|branch │ │ dt: Oct, Oct, ... │
│ (read ALL columns │ │ branch: Mum, Del,... │
│ for any query) │ │ │
└───────────────────┘ │ SELECT SUM(amt) reads│
│ ONLY the amt column │
└──────────────────────┘
Key Takeaway
Schema evolution is managing changes to data structure over time without breaking existing producers or consumers. Handle it with self-describing formats like Avro for streaming and Parquet for storage, a Schema Registry that stores versioned schemas, and compatibility rules enforced at registration. The core rules are backward compatibility where a new schema can read old data, forward compatibility where an old schema can read new data, and full compatibility which is both. The practical discipline is to add fields only with defaults, never rename in place but instead add and deprecate, and never change a field's type incompatibly.
Columnar formats like Parquet matter for a related but separate reason: analytical performance and cost. Because Parquet stores data by column, a query reads only the columns it needs (column pruning), uses per-row-group min/max statistics to skip blocks that cannot match a filter (predicate pushdown), and compresses same-type values dramatically better than mixed-type rows. Sort order makes min/max statistics effective by producing tight, non-overlapping ranges per block.
In a banking context at the National Bank of India, when the team added a channel field to transaction events to distinguish mobile from branch, they added it with a default so old consumers kept working and new consumers read it, and the Schema Registry rejected a careless change that would have added a mandatory field with no default. On the storage side, transaction history lived in sorted Parquet, so analysts totalling amounts read only the amount column and skipped irrelevant date partitions.
Parquet's embedded, evolvable schema let the table gain the new column without rewriting history, which is exactly the combination a growing bank system needs.
Remember one sentence: evolve schemas safely with a registry and compatibility rules, and use columnar Parquet so analytics reads only the columns it needs and skips irrelevant blocks.
---