Kafka-Driven Async Data Streaming: A Superior Choice Over Amazon Kinesis

Data streaming has revolutionized real-time data processing and analytics, with Apache Kafka and Amazon Kinesis emerging as prominent players in this field. In this article, we will explore why Kafka, especially when used with the Confluent Kafka framework, is often considered a superior choice compared to Amazon Kinesis. We will explore their features, and use cases, and provide advanced examples.

Kafka and Confluent Kafka

Apache Kafka, a distributed streaming platform, is designed to handle real-time data streaming and processing at scale. It gains additional capabilities when integrated with Confluent, a company that provides enterprise-ready enhancements for Kafka.

Features of Kafka and Confluent Kafka

1. Distributed Architecture: Kafka is distributed by nature, allowing for high availability and scalability. The Confluent Platform builds upon this, offering even more scalability and reliability.
2. Topic-Based Pub-Sub Model: Kafka uses a publish-subscribe model based on topics, allowing multiple consumers to subscribe to different topics, ensuring data isolation and flexibility.
3. Data Retention: Kafka retains data for a configurable period, enabling real-time and historical data analysis. Confluent’s enhancements provide fine-grained control over retention policies.

Avro Schema with Apache Kafka

Avro is a data serialization framework that works seamlessly with Apache Kafka. Avro schemas define the structure of data, allowing producers and consumers to communicate effectively. The key advantage of Avro is that it enables schema evolution, meaning you can modify schemas without breaking compatibility.

How Avro Schemas Work

1. Schema Definition: Avro schemas are defined in JSON format. They specify the data’s structure, including fields, data types, and optional default values.
2. Schema Registry: In Kafka, a schema registry stores Avro schemas. Producers and consumers can register and fetch schemas from the registry. This ensures data consistency.
3. Serialization: Data is serialized using the Avro schema before being published to a Kafka topic. This serialization allows consumers to understand and process the data, even when the schema evolves.
4. Schema Evolution: Avro supports backward and forward compatibility. This means you can add, remove, or modify fields in the schema without causing issues for existing consumers.

Confluent Kafka and Transitivity Models

Confluent Kafka, offers advanced features and tools for Kafka users. One of the features that can be vital when using Avro schemas is transitivity models. These transitivity models in Confluent Kafka refer to a set of compatibility rules and configurations that govern how schema evolution is managed within the Confluent Schema Registry. Schema evolution is the process of modifying the structure of data schemas without disrupting the ability to read and process existing data.

Transitivity Models in Confluent Kafka

Confluent Kafka introduces transitivity models to manage schema compatibility. These models define how schemas can evolve without breaking data compatibility. There are three primary models:

1. Backward Compatibility: New schema versions must be backward compatible with the previous versions. This allows producers to evolve schemas without affecting consumers. Existing fields can be modified, but you can’t remove or change existing fields in a non-backward-compatible manner.
2. Forward Compatibility: New schema versions must be forward compatible with older versions, ensuring that consumers can understand new data produced by updated schemas. This means you can add new fields or make existing fields optional.
3. Full Compatibility: New schema versions must be both backward and forward-compatible, offering the most flexibility but with stricter constraints.

An Example Avro Schema model:

{
  "type": "record",
  "name": "Employee",
  "namespace": "com.somecompany.employee.v1",
  "fields": [
    {"name": "employeeId", "type": "int"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string"}
  ]
}

• "type": "record": Indicates that this schema represents a record, which is a complex type with named fields.
• "name": "Employee": Specifies the name of the Avro record.
• "namespace": "com.somecompany.employee.v1": Provides a namespace for the record. This is a common practice to avoid naming conflicts, especially in larger schemas or when integrating with other systems. We can use a versioning mechanism to avoid ambiguity when evolving the schema to different types of integrations.
• "fields": An array defining the fields within the record. Each field is an object with a "name" and "type". In this example, we have three fields:
• "employeeId": An integer field representing the employee’s ID.
• "name": A string field representing the employee’s name.
• "email": A string field representing the employee’s email.

Choosing the Right Transitivity Model

Selecting the appropriate transitivity model depends on your specific use case and requirements. If you need to frequently evolve schemas without disrupting consumers, full compatibility may be the best choice. However, in scenarios where you want more control over schema changes, you can opt for backward or forward compatibility.

Advanced Example — Kafka and Confluent Kafka

Let’s consider a use case where you want to build a real-time event processing system using Kafka. Confluent provides a set of powerful tools and components to enhance Kafka’s capabilities. One such tool is the Confluent Schema Registry, which ensures data consistency and compatibility. we will explore how Apache Kafka, particularly when integrated with Avro schemas, excels in data streaming compared to Amazon Kinesis.

In this architecture, data producers send messages with schemas to the Kafka broker. The Schema Registry stores and manages these schemas. Consumers can retrieve the schema from the registry to understand and process the data effectively.

Here’s a simplified example of how to produce and consume messages using Spring Boot and the Confluent Kafka framework:

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.*;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.properties.schema.registry.url}")
    private String schemaRegistryUrl;

    // Producer Configuration
    @Bean
    public ProducerFactory<String, YourAvroClass> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);

        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, YourAvroClass> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    // Consumer Configuration
    @Bean
    public ConsumerFactory<String, YourAvroClass> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "your-group-id");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
        config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);

        return new DefaultKafkaConsumerFactory<>(config);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, YourAvroClass> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, YourAvroClass> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        return factory;
    }
}

// Kafka Producer
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

public void sendMessage(String message) {
    kafkaTemplate.send("my-topic", message);
}

// Kafka Consumer
@KafkaListener(topics = "my-topic", groupId = "my-group")
public void listen(String message) {
    // Process the received message
}

Amazon Kinesis

Amazon Kinesis is a cloud-based data streaming service offered by Amazon Web Services (AWS). It provides several stream processing services, such as Kinesis Data Streams, Kinesis Data Firehose, and Kinesis Data Analytics.

Features of Amazon Kinesis

1. Managed Service: Kinesis is a fully managed service, making it easy to set up and operate without the need for infrastructure management.
2. Built-in Analytics: Kinesis Data Analytics allows you to run SQL queries on streaming data, simplifying real-time data processing.
3. Integration with AWS Services: Kinesis seamlessly integrates with other AWS services like Lambda, S3, and more, enabling a comprehensive serverless architecture.

Example — Amazon Kinesis with AWS SDK

Let’s consider a scenario where you need to ingest and process data using Amazon Kinesis Data Streams with the AWS SDK in a Java-based application.

In this architecture, data is ingested into a Kinesis Data Stream and processed by Kinesis Data Analytics. The results can be stored in various AWS services for further analysis.

// AWS Kinesis Producer
AmazonKinesisClient kinesisClient = AmazonKinesisClient.builder().build();
PutRecordRequest request = new PutRecordRequest()
    .withStreamName("my-stream")
    .withData(ByteBuffer.wrap("Data to be ingested".getBytes()))
    .withPartitionKey("partition-key");

kinesisClient.putRecord(request);

// AWS Kinesis Consumer (KCL)
KinesisClientLibConfiguration config = new KinesisClientLibConfiguration(
    "my-app-name", "my-stream", new DefaultAWSCredentialsProviderChain(), "worker-id")
    .withInitialPositionInStream(InitialPositionInStream.TRIM_HORIZON);

Worker worker = new Worker.Builder()
    .recordProcessorFactory(new MyRecordProcessorFactory())
    .config(config)
    .build();

worker.run();

Kafka vs. Amazon Kinesis

Now, let’s delve into the reasons why Kafka, particularly with Confluent’s enhancements, is often considered superior to Amazon Kinesis for data streaming.

Ecosystem Maturity

Kafka has a mature and well-established ecosystem, with numerous connectors, libraries, and tools available. Confluent extends Kafka with additional features like the Schema Registry, Kafka Connect, and KSQL, making it even more versatile.

Flexibility and Control

Kafka offers greater flexibility and control over data stream processing. Users can define their data schemas, choose storage options, and fine-tune performance settings. Amazon Kinesis, while user-friendly, can be less flexible in certain scenarios.

Multi-Cloud Support

Kafka can be run on various cloud providers and on-premises, providing multi-cloud support. This flexibility is valuable for organizations seeking to avoid vendor lock-in.

Community and Support

Kafka has a large and active community, ensuring a wealth of resources and support. Confluent offers commercial support and enterprise features for businesses with specific needs.

Conclusion: While both Apache Kafka and Amazon Kinesis are solid choices for data streaming, Kafka, especially when used with Confluent’s enhancements, offers a more mature, flexible, and versatile solution. Its well-established ecosystem, flexibility, and community support make it the preferred choice for many organizations. However, the choice between Kafka and Kinesis ultimately depends on the specific requirements of your project and your familiarity with the respective technologies. An alternative approach for you if you are using Amazon Kinesis would be to consider using Amazon Kinesis Data Firehose in conjunction with AWS Glue Schema Registry. AWS Glue Schema Registry provides a schema registry service that enables you to manage and evolve Avro schemas.

References :

• What Is Amazon Kinesis Data Streams?
• Set Up Spring Boot with Confluent Cloud
• Schema Registry for Confluent Platform | Confluent Documentation
• Schema Evolution and Compatibility for Schema Registry on Confluent Platform | Confluent Documentation