Serializers & Deserializers

Overview

Serialization is a critical topic when working with Flink because it often serializes things you might not expect. Flink has two different types of serialization: Internal and External. And within each type, there are multiple formats that might be used. Depending on your choices it can have an impact on message size, flexibility, schema evolution, and more. This video will outline the different ways that Flink uses serializers and show you how to implement a few of the basics.

Topics:

• Serialization and Deserialization
• Internal and External Serialization
• POJO serialization vs Kryo serialization
• JSON Serialization

Code

A Simple POJO

public class Person {
	public String name;
	private String email;

	public Person() {}

	public String getEmail() {return email;}
	public void setEmail(String email) {this.email = email;}
}

Registring Kryo Serializers

env.getConfig().registerKryoType(MyCustomType.class);

Disabling Kryo Serialization

env.getConfig().disableGenericTypes();

JsonSerializationSchema

JsonSerializationSchema<MyClass> serializer = 
	new JsonSerializationSchema<>();

JsonDeserializationSchema

JsonDeserializationSchema<MyClass> deserializer = 
	new JsonDeserializationSchema<>(MyClass.class);

Custom Object Mapper

JsonSerializationSchema<MyClass> serializer = 
	new JsonSerializationSchema<>(() -> 
		new ObjectMapper()
			.registerModule(new JavaTimeModule())
	);

Resources

• Flink Serialization Tuning
• Data Types and Serialization
• JsonSerializationSchema
• JsonDeserializationSchema