Kafka Hadoop Integration: 2 Critical Concepts

The Kafka Hadoop Integration, or the Kafka Hadoop pipeline, is predominantly used for real-time big data analytics. Both Kafka and Hadoop are the major players in the modern data analytics landscape because they provide extended benefits when assembling a data management infrastructure from scratch.

The problem arises when engineers try to replicate continuously changing data into Kafka streams, from which the data is consumed by data lake Hadoop systems. 

To mitigate this challenge and get an overview of the solution, in this blog post, we will be exclusively talking about the Kafka Hadoop integration process and, in brief, about Apache Kafka and Hadoop. Let’s begin.

What is Kafka?

• Apache Kafka is a distributed event streaming platform. With an ability to allow applications to manage large amounts of data, Kafka is also fault-tolerant and built to scale.
• Apache Kafka’s framework is based on Java and the Publish-Subscribe Messaging system.
• The framework allows data streaming at an unprecedented rate, that too from multiple sources.
• Kafka is famous in the data community for data streaming services because it can handle Big Data with large input volumes. And, with minimum downtime and low latency, Kafka services are easy to scale up and down. 

Some Key Features of Apache Kafka

Some of Kafka’s most valuable features are as follows:

• High Scalability: The partitioned log model allows Kafka services to scale beyond a single server’s capability.
• Low Latency: Kafka services separate data streams, allowing low latency and high throughput.
• Fault-Tolerant & Durable: In Kafka, partitions are segregated and then duplicated across servers. The segregation and duplication process makes Kafka services fault-tolerant by protecting them against ad-hoc server failures like master and database failures. 
• High Extensibility: Kafka is highly accessible through various other applications, allowing developers to add more features. 

What is Hadoop?

• Hadoop, an open-source framework, is known for its efficient storage and processing speed. Hadoop can efficiently process datasets on any scale — ranging in size from gigabytes to petabytes of data.
• Hadoop uses a sizeable cluster of processors, known as parallel processing units, to store and process data, allowing it to analyze a vast set of datasets in parallel.

Hadoop contains four main modules:

• Hadoop Distributed File System (HDFS): HDFS is a distributed file system that runs on an average/ low-end hardware system. The main advantage of using HDFS is that it provides a better data throughput than traditional file systems.
• Yet Another Resource Negotiator (YARN): Managing and monitoring cluster nodes and resource usage — in short, scheduling jobs and tasks — is YARN’s responsibility.
• MapReduce: MapReduce is a governance framework that helps programs with parallel data computation. The map task takes input data and converts it into a dataset computed in key-value pairs. The output of the map task is consumed by reducing tasks to aggregate production and providing the desired result.
• Hadoop Common: Hadoop Common provides a standard Java library, making it accessible across all modules.

Kafka Hadoop Integration

In this section, we will be looking at two concepts that will help us build our Kafka Hadoop pipeline for real-time processing.

In general, we use Kafka to make this real-time processing, monitoring, and loading system that processes data through Hadoop. Let’s see how.

Hadoop Producer

A Hadoop producer introduces a connection to the Kafka producer, which helps publish data from the Hadoop cluster.

For Kafka producers, the Kafka topics are pitched as URIs, and one URI is specified below to connect to a Kafka broker:

kafka://<kafka-broker>/<kafka-topic>

The Hadoop Producer suggests two possible approaches:

Method 1: Using the Pig Script and using the Avro format to write messages

Below is a sample Pig script to illustrate how to do this:

• Kafka generally uses the Pig Scripts to write data in the Avro format. 
• Each row refers to a single message, and in order to push the data into the Kafka cluster, the AvroKafkaStorage class uses the Avro schema. 
• The Schema is first used as the first argument and then connects to the Kafka URI. 
• By using the AvroKafkaStorage producer with the Pig script, you can quickly write to multiple topics and brokers.
• Below is a sample Pig script to illustrate how to do this:

REGISTER 'path/to/avro-1.8.2.jar' 
REGISTER 'path/to/pig-avro-0.17.0.jar';


DEFINE AvroStorage org.apache.pig.piggybank.storage.AvroStorage();


-- Load your data
data = LOAD 'input_data' USING PigStorage(',') AS (field1:chararray, field2:int);


-- Store data in Kafka
STORE data INTO 'kafka://<kafka-broker>/<kafka-topic>' USING AvroStorage();

Method 2: Using the Kafka OutputFormat class for jobs

• In this method, Kafka’s OutputFormat class is utilized to publish data to the Kafka cluster. This method offers extended control over the output by using a low-level method of publishing. 
• To write a message to the Hadoop cluster, the Kafka OutputFormat Class uses the KafkaRecordWriter class. 
• For Kafka producers, we can configure Kafka Producer parameters and Kafka Broker information under the Job’s configuration settings.
• Here’s a Java example:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.kafka.hadoop.mapreduce.KafkaOutputFormat;
import org.apache.kafka.hadoop.mapreduce.KafkaRecordWriter;


public class KafkaProducerJob {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Kafka Producer Job");
        job.setJarByClass(KafkaProducerJob.class);
        job.setMapperClass(YourMapperClass.class);
        job.setOutputFormatClass(KafkaOutputFormat.class);
        // Kafka Producer configuration
        conf.set("kafka.topic", "<kafka-topic>");
        conf.set("metadata.broker.list", "<kafka-broker>");
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(YourValueClass.class);


        // Specify input and output paths
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

Learn More about how to load data from Kafka to Databricks.

Hadoop Consumer 

The process of pulling out data from the Kafka broker into HDFS is called Hadoop Consumer. The below image shows the position of the Kafka Consumer in the architecture flow process.

The Hadoop Job performs the parallel loading process that brings data from Kafka to HDFS.

Step 1: Set Up the Kafka Consumer Job

You can create a Kafka consumer job in Java using the following code:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.kafka.hadoop.mapreduce.KafkaInputFormat;


public class KafkaConsumerJob {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Kafka Consumer Job");


        job.setJarByClass(KafkaConsumerJob.class);
        job.setInputFormatClass(KafkaInputFormat.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(YourValueClass.class);


        // Kafka Consumer configuration
        conf.set("kafka.topic", "<kafka-topic>");
        conf.set("metadata.broker.list", "<kafka-broker>");


        // Specify input and output paths
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));


        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

Step 2: Pulling Data and Managing Offsets

The data from Kafka and the updated topic offsets come from the output directory. At last, the individual mappers write the offset of the last consumed message to HDFS. If the Job fails or gets restarted, each mapper restarts from the offsets stored in HDFS.

Here’s an example of how you can implement the mapper to manage offsets:

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.kafka.clients.consumer.KafkaConsumer;


public class KafkaMapper extends Mapper<NullWritable, YourValueClass, NullWritable, YourValueClass> {
    private KafkaConsumer<String, YourValueClass> consumer;


    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        // Initialize Kafka consumer
        consumer = new KafkaConsumer<>(getConsumerProperties());
        consumer.subscribe(Collections.singletonList("<kafka-topic>"));
    }


    @Override
    protected void map(NullWritable key, YourValueClass value, Context context) throws IOException, InterruptedException {
        // Process the incoming Kafka message
        context.write(key, value);
        
        // Save the offset after processing the message
        long offset = value.getOffset(); // Assuming your value has a method to get the offset
        context.getCounter("Kafka", "Offsets").increment(offset);
    }


    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        consumer.close();
    }
}

Quick Comparison

Feature	Kafka	Hadoop
Purpose	Real-time data streaming and messaging	Batch processing and data storage
Data Handling	Handles real-time data streams	Handles large volumes of batch data
Data Model	Topic-based publish/subscribe model	HDFS for file storage and MapReduce for processing
Use Cases	Log aggregation, event sourcing, stream processing	Data warehousing, analytics, ETL processes

Conclusion

In this blog post, we discussed two predominant concepts that are required to make the Kafka Hadoop Integration possible. The process makes it clear how Kafka to hdfs ingestion operates to enable real-time big data analytics.

For an in-depth analysis of big data, it is recommended to leverage Apache Kafka Hadoop integration. However, to extract, load, and transform this complex data, you would require a sizeable engineering bandwidth.

And, if you are looking for meaningful insights from Kafka data, you are just a few clicks away because Hevo Data can help.

Hevo Data, a No-code Data Pipeline, can seamlessly transfer data from a vast sea of sources, such as Apache Kafka, to a Data Warehouse or a Destination of your choice. It is a reliable, completely automated, and secure service that doesn’t require you to write any code!

Want to take Hevo for a ride? Sign Up for a 14-day free trial and simplify your Data Integration process. Check out the Hevo pricing details to understand which plan fulfills all your business needs.

Share your thoughts with us in the comments section below. Tell us about your experience completing the Kafka Hadoop Integration process!

Frequently Asked Questions