Kafka integration with Hadoop - Mini Project: Build & Apply

Kafka integration with Hadoop

📖 Scenario: You work at a company that collects real-time data streams from various sources. The data is sent through Kafka topics. Your task is to integrate Kafka with Hadoop to store and analyze this streaming data efficiently.

🎯 Goal: Build a simple pipeline that reads messages from a Kafka topic and writes them into Hadoop's HDFS for further analysis.

📋 What You'll Learn

Create a Kafka consumer configuration

Set up Hadoop file system path for data storage

Write code to consume messages from Kafka topic

Save consumed messages into HDFS files

💡 Why This Matters

🌍 Real World

Companies use Kafka to handle real-time data streams and Hadoop to store large volumes of data for analysis. Integrating them allows efficient data processing pipelines.

💼 Career

Data engineers and data scientists often build pipelines that connect streaming platforms like Kafka with big data storage systems like Hadoop.

1

Create Kafka consumer configuration

Create a dictionary called kafka_config with these exact entries: 'bootstrap.servers': 'localhost:9092', 'group.id': 'hadoop_integration_group', and 'auto.offset.reset': 'earliest'.

Hadoop

# Create kafka_config dictionary with required entries
# Your code here

Hint

Use a Python dictionary with the exact keys and values as specified.

2

Set Hadoop HDFS output path

Create a string variable called hdfs_output_path and set it to '/user/hadoop/kafka_data/'.

Hadoop

kafka_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'hadoop_integration_group',
    'auto.offset.reset': 'earliest'
}

# Set the HDFS output path
# Your code here

Hint

Assign the exact string path to the variable hdfs_output_path.

3

Consume messages from Kafka topic

Write code to create a Kafka consumer using kafka_config and subscribe to the topic 'sensor_readings'. Use a for loop with variable message to consume messages from the consumer. Append each message's value decoded as UTF-8 to a list called messages. Initialize messages as an empty list before the loop.

Hadoop

kafka_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'hadoop_integration_group',
    'auto.offset.reset': 'earliest'
}

hdfs_output_path = '/user/hadoop/kafka_data/'

# Import KafkaConsumer from kafka package
from kafka import KafkaConsumer

# Create Kafka consumer and consume messages
# Your code here

Hint

Use KafkaConsumer with the config values and subscribe to 'sensor_readings'. Decode each message value and add to messages list.

4

Save consumed messages to HDFS

Import pyarrow.fs and create a Hadoop filesystem object called hdfs. Open a file at path hdfs_output_path + 'data.txt' in write mode. Write each message from the messages list to the file, each followed by a newline. Close the file after writing. Finally, print the number of messages saved using print(f"Saved {len(messages)} messages to HDFS").

Hadoop

kafka_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'hadoop_integration_group',
    'auto.offset.reset': 'earliest'
}

hdfs_output_path = '/user/hadoop/kafka_data/'

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'sensor_readings',
    bootstrap_servers=kafka_config['bootstrap.servers'],
    group_id=kafka_config['group.id'],
    auto_offset_reset=kafka_config['auto.offset.reset']
)

messages = []
for message in consumer:
    messages.append(message.value.decode('utf-8'))
    if len(messages) >= 5:
        break

# Import pyarrow.fs and write messages to HDFS
# Your code here

Hint

Use pyarrow.fs.HadoopFileSystem to open a file stream and write each message with a newline. Then print the confirmation message.

Practice

1. What is the main purpose of integrating Kafka with Hadoop in data science?

easy

A. To move streaming data into Hadoop storage for analysis

B. To replace Hadoop's storage with Kafka's storage

C. To run machine learning models inside Kafka

D. To visualize data directly from Kafka without storage

Solution

1. Step 1: Understand Kafka and Hadoop roles

   Kafka streams data in real-time, Hadoop stores large data for analysis.

2. Step 2: Identify integration purpose

   Integration moves streaming data from Kafka into Hadoop for big data processing.

3. Final Answer:

   To move streaming data into Hadoop storage for analysis -> Option A

4. Quick Check:

   Kafka streams + Hadoop stores = Integration purpose [OK]

Hint: Kafka streams data; Hadoop stores it for analysis [OK]

Common Mistakes:

• Thinking Kafka replaces Hadoop storage
• Assuming Kafka runs ML models
• Believing visualization happens directly in Kafka

2. Which of the following is the correct command to consume messages from a Kafka topic named sensor_data?

easy

A. kafka-run-class kafka.Consumer sensor_data

B. kafka-console-producer --topic sensor_data --bootstrap-server localhost:9092

C. kafka-console-consumer --topic sensor_data --bootstrap-server localhost:9092

D. hadoop fs -cat /sensor_data

Solution

1. Step 1: Identify Kafka consumer command

   The command to read messages is kafka-console-consumer with topic and server options.

2. Step 2: Match command to options

   kafka-console-consumer --topic sensor_data --bootstrap-server localhost:9092 uses correct consumer command and parameters for topic sensor_data.

3. Final Answer:

   kafka-console-consumer --topic sensor_data --bootstrap-server localhost:9092 -> Option C

4. Quick Check:

   Consumer command with topic and server [OK]

Hint: Consumer command starts with kafka-console-consumer [OK]

Common Mistakes:

• Using producer command instead of consumer
• Confusing Hadoop fs commands with Kafka
• Using outdated or incorrect Kafka class commands

3. Given the following Flume configuration snippet to move Kafka data to HDFS:

agent.sources = kafkaSource
agent.channels = memoryChannel
agent.sinks = hdfsSink

agent.sources.kafkaSource.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.kafkaSource.kafka.bootstrap.servers = localhost:9092
agent.sources.kafkaSource.kafka.topics = sensor_data

agent.channels.memoryChannel.type = memory

agent.sinks.hdfsSink.type = hdfs
agent.sinks.hdfsSink.hdfs.path = /data/sensor
agent.sinks.hdfsSink.channel = memoryChannel

What will be the HDFS directory where Kafka data is stored?

medium

A. /data/sensor

B. /kafka/sensor_data

C. /user/flume/kafka

D. /tmp/kafka_data

Solution

1. Step 1: Locate HDFS path in sink configuration

   The sink hdfsSink has hdfs.path = /data/sensor.

2. Step 2: Confirm data storage directory

   Kafka data consumed by Flume will be stored in /data/sensor on HDFS.

3. Final Answer:

   /data/sensor -> Option A

4. Quick Check:

   HDFS sink path = /data/sensor [OK]

Hint: Check hdfs.path in sink config for storage location [OK]

Common Mistakes:

• Confusing Kafka topic name with HDFS path
• Assuming default Flume user directory
• Ignoring channel configuration

4. You set up Kafka to HDFS data transfer using Flume, but no data appears in HDFS. Which of the following is the most likely error in the Flume config?

medium

A. The memory channel type is set to file

B. The HDFS path is set correctly

C. The Flume agent name is missing

D. The Kafka topic name in kafka.topics is misspelled

Solution

1. Step 1: Check common config errors

   Misspelling Kafka topic means Flume reads no data from Kafka.

2. Step 2: Evaluate options

   Correct HDFS path and channel type do not cause no data; missing agent name causes startup error, not silent failure.

3. Final Answer:

   The Kafka topic name in kafka.topics is misspelled -> Option D

4. Quick Check:

   Wrong topic name = no data read [OK]

Hint: Check Kafka topic spelling first if no data in HDFS [OK]

Common Mistakes:

• Assuming HDFS path causes no data
• Changing channel type incorrectly
• Ignoring agent name errors

5. You want to build a pipeline that streams sensor data from Kafka into Hadoop HDFS and then runs batch analysis daily. Which combination of tools and steps is best?

hard

A. Use Kafka producer to write directly to HDFS, then run Spark streaming

B. Use KafkaSource in Flume to ingest data into HDFS, then run Hadoop MapReduce daily

C. Use Hadoop fs commands to read Kafka topics, then store in local files

D. Use Kafka consumer to write data to local disk, then upload manually to HDFS

Solution

1. Step 1: Identify streaming ingestion method

   Flume with KafkaSource is designed to stream Kafka data into HDFS efficiently.

2. Step 2: Choose batch analysis tool

   Hadoop MapReduce runs batch jobs on HDFS data daily, fitting the requirement.

3. Final Answer:

   Use KafkaSource in Flume to ingest data into HDFS, then run Hadoop MapReduce daily -> Option B

4. Quick Check:

   Flume + KafkaSource + HDFS + MapReduce = correct pipeline [OK]

Hint: Flume streams Kafka to HDFS; MapReduce runs batch jobs [OK]

Common Mistakes:

• Using Kafka producer to write directly to HDFS (not supported)
• Confusing streaming with batch tools
• Manual upload instead of automated pipeline