Swiftorial Logo
Home
Swift Lessons
Matchups
CodeSnaps
Tutorials
Career
Resources

Data Serialization in Kafka

Introduction

Data serialization is the process of transforming data structures or object states into a format that can be stored or transmitted and reconstructed later. In the context of Kafka, serialization is crucial for the efficient transmission of messages between producers and consumers. This tutorial will guide you through the fundamentals of data serialization in Kafka, covering various serialization formats and their usage.

Why Serialization?

Serialization is essential for converting complex data structures into a byte stream that can be transmitted over a network or stored in a file. In Kafka, producers serialize records before sending them to topics, and consumers deserialize records when they retrieve them. Proper serialization ensures data integrity, compatibility, and performance.

Common Serialization Formats

Several serialization formats are commonly used in Kafka:

  • JSON: A lightweight data interchange format that is easy to read and write.
  • Avro: A compact, fast, binary data format that supports rich data structures.
  • Protobuf: Google's language-neutral, platform-neutral, extensible mechanism for serializing structured data.
  • Thrift: A software framework for scalable cross-language services development.

Serialization with JSON

JSON is a popular serialization format due to its readability and ease of use. Below is an example of how to serialize and deserialize data using JSON in Kafka.

Producer Example

 
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Properties;

public class JsonProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer producer = new KafkaProducer<>(props);
        ObjectMapper objectMapper = new ObjectMapper();

        try {
            User user = new User("John", "Doe");
            String userJson = objectMapper.writeValueAsString(user);
            ProducerRecord record = new ProducerRecord<>("UserTopic", userJson);
            producer.send(record);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}

class User {
    private String firstName;
    private String lastName;

    public User(String firstName, String lastName) {
        this.firstName = firstName;
        this.lastName = lastName;
    }

    // Getters and setters...
}

Consumer Example

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Collections;
import java.util.Properties;

public class JsonConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "JsonConsumerGroup");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("UserTopic"));
        ObjectMapper objectMapper = new ObjectMapper();

        while (true) {
            ConsumerRecords records = consumer.poll(100);
            for (ConsumerRecord record : records) {
                try {
                    User user = objectMapper.readValue(record.value(), User.class);
                    System.out.println("Received User: " + user.getFirstName() + " " + user.getLastName());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

class User {
    private String firstName;
    private String lastName;

    public User() {}

    public User(String firstName, String lastName) {
        this.firstName = firstName;
        this.lastName = lastName;
    }

    public String getFirstName() {
        return firstName;
    }

    public void setFirstName(String firstName) {
        this.firstName = firstName;
    }

    public String getLastName() {
        return lastName;
    }

    public void setLastName(String lastName) {
        this.lastName = lastName;
    }
}

Serialization with Avro

Avro is a compact and efficient binary format. It is often preferred in Kafka for its schema evolution support. Here's an example of using Avro for serialization and deserialization.

Producer Example

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import java.util.Properties;

public class AvroProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
        props.put("schema.registry.url", "http://localhost:8081");

        KafkaProducer producer = new KafkaProducer<>(props);

        String userSchema = "{\"type\":\"record\",\"name\":\"User\",\"fields\":[{\"name\":\"firstName\",\"type\":\"string\"},{\"name\":\"lastName\",\"type\":\"string\"}]}";
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(userSchema);

        GenericRecord user = new GenericData.Record(schema);
        user.put("firstName", "John");
        user.put("lastName", "Doe");

        ProducerRecord record = new ProducerRecord<>("UserAvroTopic", user);

        producer.send(record);
        producer.close();
    }
}

Consumer Example

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import org.apache.avro.generic.GenericRecord;
import java.util.Collections;
import java.util.Properties;

public class AvroConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "AvroConsumerGroup");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
        props.put("schema.registry.url", "http://localhost:8081");

        KafkaConsumer consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("UserAvroTopic"));

        while (true) {
            ConsumerRecords records = consumer.poll(100);
            for (ConsumerRecord record : records) {
                GenericRecord user = record.value();
                System.out.println("Received User: " + user.get("firstName") + " " + user.get("lastName"));
            }
        }
    }
}

Serialization with Protobuf

Protobuf, developed by Google, is an efficient binary serialization format. It is language-neutral and platform-neutral, making it ideal for data interchange. Here's how to use Protobuf with Kafka.

Producer Example

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import com.example.UserProto.User;
import java.util.Properties;

public class ProtobufProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaProtobufSerializer.class.getName());

        KafkaProducer producer = new KafkaProducer<>(props);

        User user = User.newBuilder().setFirstName("John").setLastName("Doe").build();
        ProducerRecord record = new ProducerRecord<>("UserProtobufTopic", user);

        producer.send(record);
        producer.close();
    }
}

Consumer Example

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import com.example.UserProto.User;
import java.util.Collections;
import java.util.Properties;

public class ProtobufConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "ProtobufConsumerGroup");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaProtobufDeserializer.class.getName());

        KafkaConsumer consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("UserProtobufTopic"));

        while (true) {
            ConsumerRecords records = consumer.poll(100);
            for (ConsumerRecord record : records) {
                User user = record.value();
                System.out.println("Received User: " + user.getFirstName() + " " + user.getLastName());
            }
        }
    }
}

Conclusion

Data serialization is a critical aspect of Kafka's architecture, enabling efficient data transmission and storage. This tutorial covered JSON, Avro, and Protobuf serialization formats, providing examples for each. Understanding these formats and their appropriate use cases is essential for optimizing your Kafka-based data pipelines.