Apache Kafka Introduction

In this tutorial, we will discuss about the Apache Kafka key concepts with some use cases.

Introduction

Apache Kafka is a distributed event streaming platform capable of handling trillions of events a day. It was originally developed by LinkedIn and later open-sourced as part of the Apache Software Foundation. Kafka is written in Scala and Java and is used for building real-time data pipelines and streaming applications.

Key Components of Apache Kafka

Broker

A Kafka cluster consists of multiple Kafka brokers. Each broker is identified by its ID and is responsible for maintaining some portion of the data. Kafka brokers handle client requests (produce, consume) and ensure that data is reliably stored and replicated across the cluster.

Topic

Topics are logical channels to which data is sent. Each topic is split into multiple partitions, which are the basic unit of parallelism in Kafka. A topic is essentially a category or feed name to which messages are published.

Partition

A topic can have multiple partitions, which allow data to be distributed across multiple brokers for scalability and fault tolerance. Each partition is an ordered, immutable sequence of records that is continually appended to.

Producer

Producers are applications that send data to Kafka topics. Producers can choose which record to assign to which partition within the topic.

Consumer

Consumers read data from Kafka topics. Each consumer belongs to a consumer group, and Kafka ensures that each partition’s data is read by exactly one consumer within a consumer group.

Zookeeper

Kafka uses Zookeeper to manage cluster metadata, including information about topics, partitions, and broker nodes. Zookeeper helps in leader election for partitions.

Log

The Kafka log is an append-only sequence of records. Each partition in a topic corresponds to a Kafka log.

How Kafka Works

Message Production

Producers send records to a Kafka topic. Each record consists of a key, a value, and a timestamp. Producers can send records to specific partitions or let Kafka determine the partition based on the record key.

Message Consumption

Consumers read records from topics. Consumers track their read position using offsets, which are unique identifiers for each record within a partition. Consumers commit these offsets to Zookeeper or Kafka to track their progress.

Replication

Kafka replicates partitions across multiple brokers to ensure data durability and high availability. Each partition has one leader and multiple followers. The leader handles all read and write requests, while followers replicate the leader’s data.

Failover

If a broker fails, Zookeeper elects a new leader for each of the failed broker’s partitions. This ensures that the system remains available even in the presence of broker failures.

Detailed Internal Working

Producer to Broker

• Producers push data to a Kafka broker. The producer can specify the partition or rely on Kafka’s default partitioner, which uses a hash of the key to determine the partition.
• Data is written to a log file in the broker and replicated to follower brokers for durability.

Broker Storage

• Each broker stores partition data in log segments on disk. Log segments are configured with a maximum size and are periodically rolled over.
• Each log segment contains a sequence of records and is identified by a base offset.

Consumer to Broker

• Consumers poll brokers for data. They specify the topic, partition, and offset to read from.
• Consumers can be part of a consumer group, which allows for parallel data processing. Each partition is assigned to one consumer in the group.

Offset Management

• Consumers track the offset of the last read record to ensure they do not miss any records or process them multiple times.
• Offsets can be committed to Kafka or Zookeeper. Kafka’s internal __consumer_offsets topic is used for this purpose.

Replication and Failover

• Kafka ensures high availability by replicating partitions across multiple brokers.
• A leader is responsible for all reads and writes, and followers replicate the leader’s data.
• Zookeeper monitors broker health and coordinates leader election during broker failures.

Performance Considerations

Disk Throughput

• Kafka’s append-only log structure and sequential I/O enable high throughput.
• Log segments can be configured to use compression for better storage efficiency.

Network Bandwidth

• Producers and consumers batch records to optimize network usage.
• Kafka supports compression (e.g., gzip, snappy) to reduce network traffic.

Latency

• Configurable producer and consumer parameters (e.g., batch size, linger time) allow tuning for optimal latency.
• Kafka’s in-built optimizations (e.g., zero-copy transfer) reduce data copy overhead.

Example Use Case 1

Imagine a system where you want to process real-time website clickstream data to analyze user behavior.

Setting Up Kafka

• Download and install Apache Kafka.
• Start Zookeeper:

bin/zookeeper-server-start.sh config/zookeeper.properties

• Start Kafka broker:

bin/kafka-server-start.sh config/server.properties

Creating a Topic

bin/kafka-topics.sh --create --topic website_clicks --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

Producing Messages

from kafka import KafkaProducer
import json

producer = KafkaProducer(bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8'))
click_event = {'user': 'user123', 'url': '/home', 'timestamp': '2024-06-14T12:00:00Z'}
producer.send('website_clicks', value=click_event)
producer.flush()

Consuming Messages

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'website_clicks',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='website_analytics',
    value_serializer=lambda v: json.loads(v.decode('utf-8'))
)
for message in consumer:
    print(f"Received message: {message.value}")

Example Use case 2 — Real-Time Processing with Kafka Streams

Kafka Streams is a client library for building real-time applications and microservices, where the input and output data are stored in Kafka clusters.

from kafka import KafkaProducer, KafkaConsumer
import json

# Sample producer for generating click events
def produce_clicks():
    producer = KafkaProducer(bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8'))
    click_event = {'user': 'user123', 'url': '/home', 'timestamp': '2024-06-14T12:00:00Z'}
    producer.send('website_clicks', value=click_event)
    producer.flush()
# Sample consumer for processing click events
def consume_clicks():
    consumer = KafkaConsumer(
        'website_clicks',
        bootstrap_servers='localhost:9092',
        auto_offset_reset='earliest',
        enable_auto_commit=True,
        group_id='website_analytics',
        value_serializer=lambda v: json.loads(v.decode('utf-8'))
    )
    for message in consumer:
        process_click_event(message.value)
def process_click_event(click_event):
    print(f"Processing click event: {click_event}")
if __name__ == "__main__":
    produce_clicks()
    consume_clicks()

This example illustrates a simple producer that sends click events to a Kafka topic and a consumer that reads and processes these events. Kafka Streams can be used for more complex processing like filtering, joining, and aggregating events.

Example Use Case 3

In this example, we will see how a csv file data can be streamed and stored into the MySQL Database.

1. Download and install Apache Kafka.
2. Sample ecommerce_data.csv file content. Original File

date,product_id,city_id,orders
2019-12-16,1897,26,2
2019-12-16,4850,26,4
2019-12-16,2466,26,1
2019-12-16,637,26,1
2019-12-16,3497,26,184

3. Sample Producer code

import time
from kafka import KafkaProducer
import os
import json

bootstrap_servers = ['localhost:9092']
topicName = 'producer-consumer-demo'
producer = KafkaProducer(bootstrap_servers = bootstrap_servers, retries = 5,value_serializer=lambda m: json.dumps(m).encode('ascii'))
with open('ecommerce_data.csv') as f:
    head = [val.strip('\n') for val in next(f).split(',')]
    for line in f:
        content = dict(zip(head, [val.strip('\n') for val in line.split(',')]))
        print(content)
        ack = producer.send(topicName, content)
        metadata = ack.get()
        #time.sleep(2)

4. Install mysql

5. Create DB, and required Table

CREATE DATABASE Kafka;
use Kafka;
CREATE TABLE orders(date VARCHAR(20), product_id INTEGER(20), city_id INTEGER(20), orders INTEGER(20));

6. Sample Consumer Code

from kafka import KafkaConsumer
import sys
import json
import mysql.connector

mydb = mysql.connector.connect(
  host="localhost",
  user="root",
  password="XXXXXXXX",
  database="Kafka"
)
mycursor = mydb.cursor()
bootstrap_servers = ['localhost:9092']
topicName = 'producer-consumer-demo'
consumer = KafkaConsumer(topicName,bootstrap_servers = bootstrap_servers, auto_offset_reset = 'latest')
try:
    for message in consumer:
        msg = json.loads(message.value)
        sql = "INSERT INTO orders(date, product_id, city_id, orders) VALUES (%s, %s, %s, %s)"
        val = (msg['date'], int(msg['product_id']), int(msg['city_id']), int(msg['orders']))
        print(val)

        mycursor.execute(sql, val)
        mydb.commit()
    # disconnecting from server
    mydb.close()
except KeyboardInterrupt:
    # disconnecting from server
    mydb.close()
    sys.exit()

7. Data inserted into Table after producer and consumer execution

Data Inserted into mysql using Python & Kafka

References:

If you want to gain more skills on Apache Kafka, you can get from the below resources.

Apache Kafka Quick Start Guide

Kafka: The Definitive Guide

Apache Kafka Cookbook

A Guide to Apache Spark, Mesos, Akka, Cassandra, and Kafka

Kafka in Action