Definition
- TypeInformation – Flink's type system that inspects each data type at job compile time to select the best TypeSerializer.
- Built-in serializers – Highly optimized serializers for primitives, basic types, arrays, Tuples, Rows, and Avro-generated classes that write compact binary with no per-record reflection.
- PojoSerializer – A serializer for well-formed POJOs (public class, public no-arg constructor, public or accessor-reachable fields) that serializes field by field efficiently and supports state schema evolution.
- Kryo fallback – The generic serializer used when Flink cannot analyze a type; it works for almost anything but is slower, produces larger output, and does not support state evolution.
- State schema evolution – The ability to add or remove fields from a state class and have Flink migrate existing state when restoring from a savepoint.
Every time a transaction record at the National Bank of India moves between departments, travels over the wire, or gets saved into the vault of checkpointed state, it must be packed into a compact form. Flink is like a specialist packer who has pre-built, perfectly sized boxes for standard items like numbers and well-labeled forms. But if you hand the packer an oddly shaped object it has never seen, it falls back to slow general-purpose wrapping (Kryo). If you design your transaction form properly, with a clear blank template and labeled fields, the packer uses fast boxes and can even adapt when you later add a new field.
Understanding the Concepts
What is Flink's type system?
Flink serializes data in three hot paths: when records pass between operators or across the network, when state is read and written, and when checkpoints are taken. Instead of using Java's built-in Serializable mechanism, Flink analyzes each type through its TypeInformation system at compile time and picks an optimized TypeSerializer.
package com.bank.model;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
public class TypeInformationDemo {
public static void main(String[] args) {
// Flink knows how to serialize these efficiently
TypeInformation<String> stringType = Types.STRING;
TypeInformation<Integer> intType = Types.INT;
// For custom classes, Flink inspects the class
TypeInformation<Transaction> txnType =
TypeInformation.of(Transaction.class);
System.out.println("Type: " + txnType);
// If POJO: "PojoType<Transaction, fields=[...]>"
// If Kryo: "GenericType<Transaction>"
}
}
What makes a class a POJO in Flink?
A class is recognized as a POJO when it is public and standalone, has a public no-argument constructor, and every non-transient field is either public or accessible through conforming public getters and setters. When these conditions hold, Flink builds a PojoSerializer.
package com.bank.model;
import java.math.BigDecimal;
// CORRECT POJO: Flink uses PojoSerializer
public class Transaction {
private String accountId; // private with getter/setter
private BigDecimal amount;
private long eventTime;
public Transaction() {} // public no-arg constructor
public String getAccountId() { return accountId; }
public void setAccountId(String accountId) {
this.accountId = accountId;
}
public BigDecimal getAmount() { return amount; }
public void setAmount(BigDecimal amount) {
this.amount = amount;
}
public long getEventTime() { return eventTime; }
public void setEventTime(long eventTime) {
this.eventTime = eventTime;
}
}
package com.bank.model;
import java.math.BigDecimal;
// WRONG: NOT a POJO (no no-arg constructor)
// Flink falls back to Kryo (slower, no state evolution)
public class BadTransaction {
private final String accountId;
private final BigDecimal amount;
// Only an all-args constructor, no no-arg constructor!
public BadTransaction(String accountId, BigDecimal amount) {
this.accountId = accountId;
this.amount = amount;
}
public String getAccountId() { return accountId; }
public BigDecimal getAmount() { return amount; }
// Missing setters for private fields!
}
How is POJO serialization different from Kryo?
| Feature | PojoSerializer | Kryo Serializer |
|---|---|---|
| Speed | Fast (field-by-field, no reflection per record) | Slower (generic framework) |
| Output size | Compact | Larger |
| State evolution | Supported (add/remove fields) | Not supported |
| Requirements | Public class, no-arg constructor, accessors | Any class |
| Registration | Automatic | Manual registration for best performance |
| Detection | Logged as PojoType | Logged as GenericType (warning) |
// Check which serializer Flink chose
// In logs, look for:
// INFO: Type of field 'transaction' is PojoType<Transaction, ...>
// --> GOOD: PojoSerializer
// WARN: Type of field 'transaction' is GenericType<Transaction>
// --> BAD: Kryo fallback (slower, no state evolution)
Why does serializer choice affect performance?
PojoSerializer writes compact binary with no per-record reflection, producing smaller records and smaller checkpoints. This means less network traffic during shuffles, less disk I/O during checkpoints, and faster recovery.
package com.bank.benchmark;
public class SerializationPerformance {
// Benchmark comparison (approximate):
//
// Transaction record with 5 fields:
// PojoSerializer: ~80 bytes, ~200ns per serialize
// KryoSerializer: ~150 bytes, ~500ns per serialize
//
// At 100,000 records/second:
// Pojo: 8 MB/s network, 20ms CPU
// Kryo: 15 MB/s network, 50ms CPU
//
// Checkpoint with 10M keyed states:
// Pojo: ~800 MB checkpoint size
// Kryo: ~1500 MB checkpoint size
// Kryo checkpoints take nearly twice as long
}
How does state schema evolution work?
The PojoSerializer and Avro serializer support changing the schema of state across job versions. You can add a new field, remove a field, and Flink migrates existing state when restoring from a savepoint.
package com.bank.model;
import java.math.BigDecimal;
// Version 1: original Transaction class
public class Transaction {
private String accountId;
private BigDecimal amount;
private long eventTime;
// getters, setters, no-arg constructor
}
// Version 2: added riskScore field
// Flink migrates existing state: old records get riskScore = null
public class Transaction {
private String accountId;
private BigDecimal amount;
private long eventTime;
private Double riskScore; // NEW FIELD
// getters, setters, no-arg constructor
}
// Restore from savepoint:
// Old state with 3 fields --> new class with 4 fields
// Flink PojoSerializer handles the migration automatically
// Kryo serializer would FAIL: state not restorable!
How do you disable Kryo fallback during development?
Disabling the generic Kryo fallback makes Flink throw an error when it encounters a non-POJO type, surfacing the problem early rather than silently degrading performance.
package com.bank.streaming;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class DisableKryoDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
// Disable Kryo fallback during development
env.getConfig().disableGenericTypes();
// Now if any type is not a POJO, Flink throws:
// UnsupportedOperationException:
// Generic types have been disabled and type
// BadTransaction is not a POJO
// This catches the problem at development time!
env.execute("NBI Strict Serialization Demo");
}
}
What are common mistakes with serialization?
The most common mistake is using a class without a no-arg constructor as keyed state, causing a silent Kryo fallback that halves throughput and prevents state evolution. Another mistake is using Java records (which lack no-arg constructors) as Flink state types.
// WRONG: Java record as state type (no no-arg constructor)
// public record TransactionRecord(String accountId, BigDecimal amount) {}
// Flink cannot use PojoSerializer for records!
// Falls back to Kryo: slower, no state evolution
// CORRECT: use a proper POJO class for state
public class TransactionPojo {
private String accountId;
private BigDecimal amount;
public TransactionPojo() {} // no-arg constructor
// getters and setters
}
// For Avro: use Avro-generated classes
// They have built-in serializers with full schema evolution
What should a senior engineer know about serialization?
A senior engineer should verify domain classes are recognized as POJOs, prefer POJOs or Avro for long-lived evolving state, disable generic types during development, and mark genuinely transient fields as transient so they are skipped during serialization.
package com.bank.model;
import java.math.BigDecimal;
public class SeniorTransaction {
private String accountId;
private BigDecimal amount;
private long eventTime;
// Transient: not serialized, not part of state
private transient String cachedDisplay;
public SeniorTransaction() {} // no-arg constructor required
// All getters and setters for non-transient fields
public String getAccountId() { return accountId; }
public void setAccountId(String v) { this.accountId = v; }
public BigDecimal getAmount() { return amount; }
public void setAmount(BigDecimal v) { this.amount = v; }
public long getEventTime() { return eventTime; }
public void setEventTime(long v) { this.eventTime = v; }
}
Senior serialization checklist:
1. Verify type is detected as PojoType (not GenericType) in logs
2. Use disableGenericTypes() in dev to catch problems early
3. Use POJO or Avro for state that must evolve across versions
4. Mark transient fields: not serialized, not in checkpoints
5. For Avro: use schema registry for formal versioning
6. Monitor: checkpoint size as proxy for serialization efficiency
Visual Overview
Flink Type System and Serializer Selection
┌─────────────────────────────────────────────────────┐
│ Type Analysis at Compile Time │
│ │
│ Class ──▶ TypeInformation ──▶ TypeSerializer │
│ │
│ ┌──────────────┐ ┌──────────────────────────────┐ │
│ │ Primitives │ │ Built-in: IntSerializer, │ │
│ │ String, int │──│ StringSerializer, etc. │ │
│ └──────────────┘ └──────────────────────────────┘ │
│ ┌──────────────┐ ┌──────────────────────────────┐ │
│ │ Tuple, Row │ │ Built-in: TupleSerializer, │ │
│ │ Avro class │──│ AvroSerializer │ │
│ └──────────────┘ └──────────────────────────────┘ │
│ ┌──────────────┐ ┌──────────────────────────────┐ │
│ │ Well-formed │ │ PojoSerializer │ │
│ │ POJO class │──│ (fast, compact, evolvable) │ │
│ └──────────────┘ └──────────────────────────────┘ │
│ ┌──────────────┐ ┌──────────────────────────────┐ │
│ │ Unknown / │ │ KryoSerializer (FALLBACK) │ │
│ │ non-POJO │──│ (slow, large, not evolvable) │ │
│ └──────────────┘ └──────────────────────────────┘ │
└─────────────────────────────────────────────────────┘
POJO Requirements:
┌─────────────────────────────────────┐
│ 1. Public class │
│ 2. Public no-arg constructor │
│ 3. All non-transient fields: │
│ - Public OR │
│ - Have public getter + setter │
└─────────────────────────────────────┘
All met? ──▶ PojoSerializer (fast)
Any missing? ──▶ KryoSerializer (slow)
Key Takeaway
Flink serializes every record that crosses operators, gets shuffled over the network, or is stored in state and checkpoints. It does this through its own type system rather than Java's built-in serialization. At compile time Flink analyzes each type through TypeInformation and picks a TypeSerializer. For primitives, Tuples, Rows, Avro classes, and well-formed POJOs it has optimized serializers that write compact binary with no per-record reflection. For any type it cannot analyze, it falls back to the generic Kryo serializer.
The reason defining POJOs correctly matters comes down to two concrete benefits. First, performance: PojoSerializer produces compact records and smaller checkpoints, meaning less network traffic and faster recovery. Second, state schema evolution: the PojoSerializer lets you add or remove fields across job versions and Flink migrates existing state on restore, whereas Kryo-serialized state generally cannot evolve, making old savepoints unrestorable.
At the National Bank of India, a transaction class used as keyed state had only an all-arguments constructor and private fields with no accessors, so Flink had silently fallen back to Kryo. Throughput was low and checkpoints were surprisingly large. Once I added a public no-arg constructor and proper getters, Flink used the PojoSerializer, records and checkpoints shrank, and throughput rose. Later, a regulatory change required a new field; because the class was now a proper POJO, Flink migrated the existing state cleanly on restore.
The senior rule is to always verify your domain classes are recognized as POJOs, disable the generic Kryo fallback during development, and use POJO or Avro for any state that must evolve.
Remember one sentence: Define POJOs with a public no-arg constructor and accessors so Flink uses its fast PojoSerializer instead of falling back to slow Kryo, enabling both performance and state schema evolution.
---