Customizing Serializers for Flink State

Customizing Flink state serialization is an advanced technique that can improve compatibility and performance in complex scenarios.

What a State Serializer Does

These two lines create an Integer ValueState:

1
// 创建 Integer 类型的 ValueState
2
ValueStateDescriptor<Integer> valueState = new ValueStateDescriptor<>("value", Types.INT);
3
ValueState<Integer> state = getRuntimeContext().getState(valueState);

Passing Types.INT tells Flink that the state holds Integers, for which it creates an IntSerializer. That serializer has two roles:

• During snapshots, serialize Integer state values into binary data for persistent storage.
• During restore, deserialize stored binary data back into Integer objects.

A state serializer therefore defines the state’s binary format. Customizing it can solve common problems such as:

1. High overhead for complex objects, such as Kryo writing redundant className information.
2. Failed recovery after fields are added to or removed from state objects.

Basic Concepts

A complete state serializer consists of a Serializer and SerializerSnapshot:

• Serializer defines serialization and deserialization for a type.
• SerializerSnapshot defines how serializer metadata is serialized as part of the snapshot.
• TypeSerializerSchemaCompatibility describes compatibility between the user-defined TypeSerializer after restart and the one in the snapshot.

The official Custom Serialization for Managed State documentation explains these concepts in more detail.

Implementing a Serializer

We will examine DecimalDataSerializer, which serializes DecimalData.

First, the DecimalData structure:

1
public final class DecimalData implements Comparable<DecimalData> {
2
  final int precision;   // 精度（字段长度）
3
  final int scale;  // 范围（小数的位数）
4
  final long longVal;  // 精度较小时，选择用 long 值表示
5
  BigDecimal decimalVal;   // 精度较大时，选择用 java.math.BigDecimal 表示
6

7
  // 判断是否超出最大精度（即是否可以使用 long 值表示）
8
  public boolean isCompact() {
9
    return precision <= MAX_COMPACT_PRECISION;
10
  }
11

12
  // 转换成 long 值
13
  public long toUnscaledLong() {
14
    if (isCompact()) {
15
      return longVal;
16
    } else {
17
      return toBigDecimal().unscaledValue().longValueExact();
18
    }
19
  }
20

21
  // 转换成 bytes
22
  public byte[] toUnscaledBytes() {
23
    return toBigDecimal().unscaledValue().toByteArray();
24
  }
25
}

Its corresponding serializer:

1
public final class DecimalDataSerializer extends TypeSerializer<DecimalData> {
2
  private final int precision;
3
  private final int scale;
4

5
  // 构造函数，初始化 precision 和 scale
6
  public DecimalDataSerializer(int precision, int scale) {
7
    this.precision = precision;
8
    this.scale = scale;
9
  }
10

11
  @Override
12
  public void serialize(DecimalData record, DataOutputView target) throws IOException {
13
    if (DecimalData.isCompact(precision)) {
14
      // 当前精度小，使用 long 值表示，写出 long 值
15
      assert record.isCompact();
16
      target.writeLong(record.toUnscaledLong());
17
    } else {
18
      // 当前精度大，使用 BigDecimal 表示
19
      byte[] bytes = record.toUnscaledBytes();
20
      target.writeInt(bytes.length);
21
      target.write(bytes);
22
    }
23
  }
24

25
  @Override
26
  public DecimalData deserialize(DataInputView source) throws IOException {
27
    if (DecimalData.isCompact(precision)) {
28
      // 当前精度小，读取 long 值，初始化 DecimalData
29
      long longVal = source.readLong();
30
      return DecimalData.fromUnscaledLong(longVal, precision, scale);
31
    } else {
32
      // 当前精度大，读取 bytes，初始化 DecimalData
33
      int length = source.readInt();
34
      byte[] bytes = new byte[length];
35
      source.readFully(bytes);
36
      return DecimalData.fromUnscaledBytes(bytes, precision, scale);
37
    }
38
  }
39
}

The code shows two core methods:

• serialize: Converts DecimalData into binary data written to DataOutputView.
• deserialize: Reads binary data sequentially from DataInputView and reconstructs DecimalData.

You do not need to decide when these methods run. Whether used for RPC or a local/remote filesystem, the corresponding OutputStream and InputStream are wrapped in DataOutputView and DataInputView.

Supporting Multiple Versions

If precision or scale changes after a job modification, DecimalData.isCompact(precision) may produce a different result, making existing bytes impossible to deserialize. What must change to build a serializer that tolerates such modifications?

Include Version Information

To allow precision and scale to change, modify the serialize and deserialize methods as follows:

1
  int currentVersion = 2;
2

3
  @Override
4
  public void serialize(DecimalData record, DataOutputView target) throws IOException {
5
    target.writeInt(currentVersion); // 新增版本信息
6
    // ... 和之前一致
7
  }
8

9
  @Override
10
  public DecimalData deserialize(DataInputView source) throws IOException {
11
    int version = source.readInt();
12
    if (version == currentVersion) {
13
      return deserializeCurrentVersion(source);
14
    } else {
15
      return deserializeVersion1(source);
16
    }
17
  }
18

19
  // 解析当前版本的数据结构
20
  private DecimalData deserializeCurrentVersion(DataInputView source) {
21
    // 使用当前版本的 precision 和 scale 进行解析
22
  }
23

24
  // 解析 Version 1 版本的数据结构
25
  private DecimalData deserializeVersion1(DataInputView source) {
26
    // 使用 Version 1 中的 precision 和 scale 进行解析并转化成当前结构的数据
27
  }

Adjust Compatibility Checks

After precision or scale changes, the snapshot’s DecimalDataSerializer no longer matches the newly created one. Flink checks these properties and throws TypeSerializerSchemaCompatibility.incompatible() on a mismatch. With version information included, we can instead return TypeSerializerSchemaCompatibility.compatibleAsIs().

There are four compatibility outcomes:

• COMPATIBLE_AS_IS: Compatible; use the new user-defined serializer going forward.
• COMPATIBLE_AFTER_MIGRATION: Compatible after rewriting state: deserialize with the snapshot’s serializer, then serialize with the new one.
• COMPATIBLE_WITH_RECONFIGURED_SERIALIZER: Compatible; return a reconfigured serializer for future use.
• INCOMPATIBLE: Incompatible; the job throws an exception and exits.

References

1. DecimalDataSerializer.java
2. Custom Serialization for Managed State

Custom serializers are common in platform applications, especially DataStream-based frameworks. Users store business state whose fields frequently change. Flink’s built-in implementations, such as IntSerializer and DecimalDataSerializer, provide useful examples.