Kafka Serialization of an object

Kafka

Serialization

Object Serialization

Data Processing

Programming

Kafka Serialization of an object

Master System Design with Codemia

Enhance your system design skills with over 120 practice problems, detailed solutions, and hands-on exercises.

Start Practicing Learn More

Apache Kafka is a powerful distributed streaming platform capable of handling trillions of events a day. Initially conceived as a message queue, Kafka is based on the concept of a distributed commit log. As you send data into Kafka via producers and read data via consumers, one critical aspect to ensure efficient data handling is serialization. Serialization in Kafka plays a pivotal role in how data is stored, transmitted, and processed in real parts of your application.

What is Serialization?

Serialization is the process of converting an object into a stream of bytes to store the object or transmit it to memory, a database, or a file. Its main purpose is to save the state of an object in order to be able to recreate it when needed. The reverse process is called deserialization.

Importance of Serialization in Kafka

In Kafka, producers send messages that consist of a key and a value and both keys and values can consist of complex objects. Kafka, being a byte-oriented system, does not understand Java, Python, or any language-specific objects. Therefore, there is a need for serialization to convert these objects into bytes before they are sent to Kafka topics.

How Kafka Handles Serialization

Kafka uses serializers in the producer to convert the objects to bytes and deserializers in the consumer to rebuild the objects from these byte arrays. Kafka provides default serializers and deserializers for simple data types like String, Integer, and a few others. For any other data type (e.g., custom objects), you must provide a custom serializer when configuring your producer and consumer.

Implementing Custom Serialization

Let's consider a Java example where you have a User object with id, name, and email fields. To allow Kafka to handle this user object correctly, custom serialization and deserialization logic are needed.

Custom Serializer

Implement a class UserSerializer that implements the Kafka Serializer interface.

java

1import org.apache.kafka.common.serialization.Serializer;
2import java.nio.ByteBuffer;
3import java.util.Map;
4
5public class UserSerializer implements Serializer<User> {
6    @Override
7    public void configure(Map<String, ?> configs, boolean isKey) {
8        // Configuration, not needed for this simple example.
9    }
10
11    @Override
12    public byte[] serialize(String topic, User data) {
13        int sizeOfName;
14        int sizeOfEmail;
15        byte[] serializedName;
16        byte[] serializedEmail;
17
18        try {
19            if (data == null)
20                return null;
21
22            serializedName = data.getName().getBytes("UTF-8");
23            sizeOfName = serializedName.length;
24            serializedEmail = data.getEmail().getBytes("UTF-8");
25            sizeOfEmail = serializedEmail.length;
26
27            ByteBuffer buffer =
28                ByteBuffer.allocate(4 + sizeOfName + 4 + sizeOfEmail);
29            buffer.putInt(sizeOfName);
30            buffer.put(serializedName);
31            buffer.putInt(sizeOfEmail);
32            buffer.put(serializedEmail);
33
34            return buffer.array();
35        } catch (Exception e) {
36            throw new IllegalArgumentException("Error in serialization", e);
37        }
38    }
39
40    @Override
41    public void close() {
42        // Nothing to close, but you can use this method if you need to clean up any resources.
43    }
44}

Custom Deserializer

Implement a class UserDeserializer that implements the Kafka Deserializer interface.

java

1import org.apache.kafka.common.serialization.Deserializer;
2import java.nio.ByteBuffer;
3import java.util.Map;
4
5public class UserDeserializer implements Deserializer<User> {
6    @Override
7    public void configure(Map<String, ?> configs, boolean isKey) {
8        // Configuration, not needed for this simple example.
9    }
10
11    @Override
12    public User deserialize(String topic, byte[] data) {
13        if (data == null)
14            return null;
15
16        ByteBuffer buffer = ByteBuffer.wrap(data);
17        int sizeOfName = buffer.getInt();
18        byte[] nameBytes = new byte[sizeOfName];
19        buffer.get(nameBytes);
20        String deserializedName = new String(nameBytes, "UTF-8");
21        int sizeOfEmail = buffer.getInt();
22        byte[] emailBytes = new byte[sizeOfEmail];
23        buffer.get(emailBytes);
24        String deserializedEmail = new String(emailBytes, "UTF-8");
25
26        return new User(deserializedName, deserializedEmail);
27    }
28
29    @Override
30    public void close() {
31        // Nothing to close, but you can use this method if needed.
32    }
33}

Serialization Configuration in Kafka Producer and Consumer

When configuring your Kafka producer or consumer, set the serializer or deserializer in the properties:

java

1// For producers
2Properties props = new Properties();
3props.put("bootstrap.servers", "localhost:9092");
4props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
5props.put("value.serializer", "com.example.UserSerializer");
6
7Producer<String, User> producer = new KafkaProducer<>(props);
8
9// For consumers
10Properties props = new Properties();
11props.put("bootstrap.servers", "localhost:9092");
12props.put("group.id", "test-group");
13props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
14props.put("value.deserializer", "com.example.UserDeserializer");
15
16Consumer<String, User> consumer = new KafkaConsumer<>(props);

Summary

Here is a summary table of key points in Kafka serialization:

Aspect	Description
What is it?	Conversion of an object into a byte stream for storage or transmission.
Importance	Essential for Kafka to store and transmit objects as it operates on byte arrays.
Default Support	Kafka provides built-in serializers for basic types like String and Integer.
Custom Handling	Custom serializers/deserializers are needed for any non-supported object types.

Conclusion

Effective serialization ensures efficient data handling in Kafka by maintaining the integrity and usability of transmitted data. When dealing with complex data types or custom objects, having robust serialization/deserialization logic is crucial for Kafka systems to operate correctly and efficiently.