Apache Kafka - Guide rapide

Dans le Big Data, un énorme volume de données est utilisé. Concernant les données, nous avons deux défis principaux: le premier est de savoir comment collecter un grand volume de données et le second est d'analyser les données collectées. Pour surmonter ces défis, vous devez avoir besoin d'un système de messagerie.

Kafka est conçu pour les systèmes distribués à haut débit. Kafka a tendance à très bien fonctionner en remplacement d'un courtier de messages plus traditionnel. Par rapport à d'autres systèmes de messagerie, Kafka offre un meilleur débit, un partitionnement intégré, une réplication et une tolérance aux pannes inhérente, ce qui en fait un outil idéal pour les applications de traitement de messages à grande échelle.

Qu'est-ce qu'un système de messagerie?

Un système de messagerie est responsable du transfert des données d'une application à une autre, afin que les applications puissent se concentrer sur les données, sans se soucier de la façon de les partager. La messagerie distribuée est basée sur le concept de mise en file d'attente fiable des messages. Les messages sont mis en file d'attente de manière asynchrone entre les applications clientes et le système de messagerie. Deux types de modèles de messagerie sont disponibles: l'un est point à point et l'autre est un système de messagerie de publication-abonnement (pub-sub). La plupart des modèles de messagerie suiventpub-sub.

Point to Point Messaging System

In a point-to-point system, messages are persisted in a queue. One or more consumers can consume the messages in the queue, but a particular message can be consumed by a maximum of one consumer only. Once a consumer reads a message in the queue, it disappears from that queue. The typical example of this system is an Order Processing System, where each order will be processed by one Order Processor, but Multiple Order Processors can work as well at the same time. The following diagram depicts the structure.

Publish-Subscribe Messaging System

In the publish-subscribe system, messages are persisted in a topic. Unlike point-to-point system, consumers can subscribe to one or more topic and consume all the messages in that topic. In the Publish-Subscribe system, message producers are called publishers and message consumers are called subscribers. A real-life example is Dish TV, which publishes different channels like sports, movies, music, etc., and anyone can subscribe to their own set of channels and get them whenever their subscribed channels are available.

What is Kafka?

Apache Kafka is a distributed publish-subscribe messaging system and a robust queue that can handle a high volume of data and enables you to pass messages from one end-point to another. Kafka is suitable for both offline and online message consumption. Kafka messages are persisted on the disk and replicated within the cluster to prevent data loss. Kafka is built on top of the ZooKeeper synchronization service. It integrates very well with Apache Storm and Spark for real-time streaming data analysis.

Benefits

Following are a few benefits of Kafka −

• Reliability − Kafka is distributed, partitioned, replicated and fault tolerance.

• Scalability − Kafka messaging system scales easily without down time..

• Durability − Kafka uses Distributed commit log which means messages persists on disk as fast as possible, hence it is durable..

• Performance − Kafka has high throughput for both publishing and subscribing messages. It maintains stable performance even many TB of messages are stored.

Kafka is very fast and guarantees zero downtime and zero data loss.

Use Cases

Kafka can be used in many Use Cases. Some of them are listed below −

• Metrics − Kafka is often used for operational monitoring data. This involves aggregating statistics from distributed applications to produce centralized feeds of operational data.

• Log Aggregation Solution − Kafka can be used across an organization to collect logs from multiple services and make them available in a standard format to multiple con-sumers.

• Stream Processing − Popular frameworks such as Storm and Spark Streaming read data from a topic, processes it, and write processed data to a new topic where it becomes available for users and applications. Kafka’s strong durability is also very useful in the context of stream processing.

Need for Kafka

Kafka is a unified platform for handling all the real-time data feeds. Kafka supports low latency message delivery and gives guarantee for fault tolerance in the presence of machine failures. It has the ability to handle a large number of diverse consumers. Kafka is very fast, performs 2 million writes/sec. Kafka persists all data to the disk, which essentially means that all the writes go to the page cache of the OS (RAM). This makes it very efficient to transfer data from page cache to a network socket.

Before moving deep into the Kafka, you must aware of the main terminologies such as topics, brokers, producers and consumers. The following diagram illustrates the main terminologies and the table describes the diagram components in detail.

In the above diagram, a topic is configured into three partitions. Partition 1 has two offset factors 0 and 1. Partition 2 has four offset factors 0, 1, 2, and 3. Partition 3 has one offset factor 0. The id of the replica is same as the id of the server that hosts it.

Assume, if the replication factor of the topic is set to 3, then Kafka will create 3 identical replicas of each partition and place them in the cluster to make available for all its operations. To balance a load in cluster, each broker stores one or more of those partitions. Multiple producers and consumers can publish and retrieve messages at the same time.

S.No	Components and Description
1	Topics A stream of messages belonging to a particular category is called a topic. Data is stored in topics. Topics are split into partitions. For each topic, Kafka keeps a mini-mum of one partition. Each such partition contains messages in an immutable ordered sequence. A partition is implemented as a set of segment files of equal sizes.
2	Partition Topics may have many partitions, so it can handle an arbitrary amount of data.
3	Partition offset Each partitioned message has a unique sequence id called as offset.
4	Replicas of partition Replicas are nothing but backups of a partition. Replicas are never read or write data. They are used to prevent data loss.
5	Brokers Brokers are simple system responsible for maintaining the pub-lished data. Each broker may have zero or more partitions per topic. Assume, if there are N partitions in a topic and N number of brokers, each broker will have one partition. Assume if there are N partitions in a topic and more than N brokers (n + m), the first N broker will have one partition and the next M broker will not have any partition for that particular topic. Assume if there are N partitions in a topic and less than N brokers (n-m), each broker will have one or more partition sharing among them. This scenario is not recommended due to unequal load distri-bution among the broker.
6	Kafka Cluster Kafka’s having more than one broker are called as Kafka cluster. A Kafka cluster can be expanded without downtime. These clusters are used to manage the persistence and replication of message data.
7	Producers Producers are the publisher of messages to one or more Kafka topics. Producers send data to Kafka brokers. Every time a producer pub-lishes a message to a broker, the broker simply appends the message to the last segment file. Actually, the message will be appended to a partition. Producer can also send messages to a partition of their choice.
8	Consumers Consumers read data from brokers. Consumers subscribes to one or more topics and consume published messages by pulling data from the brokers.
9	Leader Leader is the node responsible for all reads and writes for the given partition. Every partition has one server acting as a leader.
10	Follower Node which follows leader instructions are called as follower. If the leader fails, one of the follower will automatically become the new leader. A follower acts as normal consumer, pulls messages and up-dates its own data store.

Take a look at the following illustration. It shows the cluster diagram of Kafka.

The following table describes each of the components shown in the above diagram.

S.No	Components and Description
1	Broker Kafka cluster typically consists of multiple brokers to maintain load balance. Kafka brokers are stateless, so they use ZooKeeper for maintaining their cluster state. One Kafka broker instance can handle hundreds of thousands of reads and writes per second and each bro-ker can handle TB of messages without performance impact. Kafka broker leader election can be done by ZooKeeper.
2	ZooKeeper ZooKeeper is used for managing and coordinating Kafka broker. ZooKeeper service is mainly used to notify producer and consumer about the presence of any new broker in the Kafka system or failure of the broker in the Kafka system. As per the notification received by the Zookeeper regarding presence or failure of the broker then pro-ducer and consumer takes decision and starts coordinating their task with some other broker.
3	Producers Producers push data to brokers. When the new broker is started, all the producers search it and automatically sends a message to that new broker. Kafka producer doesn’t wait for acknowledgements from the broker and sends messages as fast as the broker can handle.
4	Consumers Since Kafka brokers are stateless, which means that the consumer has to maintain how many messages have been consumed by using partition offset. If the consumer acknowledges a particular message offset, it implies that the consumer has consumed all prior messages. The consumer issues an asynchronous pull request to the broker to have a buffer of bytes ready to consume. The consumers can rewind or skip to any point in a partition simply by supplying an offset value. Consumer offset value is notified by ZooKeeper.

As of now, we discussed the core concepts of Kafka. Let us now throw some light on the workflow of Kafka.

Kafka is simply a collection of topics split into one or more partitions. A Kafka partition is a linearly ordered sequence of messages, where each message is identified by their index (called as offset). All the data in a Kafka cluster is the disjointed union of partitions. Incoming messages are written at the end of a partition and messages are sequentially read by consumers. Durability is provided by replicating messages to different brokers.

Kafka provides both pub-sub and queue based messaging system in a fast, reliable, persisted, fault-tolerance and zero downtime manner. In both cases, producers simply send the message to a topic and consumer can choose any one type of messaging system depending on their need. Let us follow the steps in the next section to understand how the consumer can choose the messaging system of their choice.

Workflow of Pub-Sub Messaging

Following is the step wise workflow of the Pub-Sub Messaging −

• Producers send message to a topic at regular intervals.

• Kafka broker stores all messages in the partitions configured for that particular topic. It ensures the messages are equally shared between partitions. If the producer sends two messages and there are two partitions, Kafka will store one message in the first partition and the second message in the second partition.

• Consumer subscribes to a specific topic.

• Once the consumer subscribes to a topic, Kafka will provide the current offset of the topic to the consumer and also saves the offset in the Zookeeper ensemble.

• Consumer will request the Kafka in a regular interval (like 100 Ms) for new messages.

• Once Kafka receives the messages from producers, it forwards these messages to the consumers.

• Consumer will receive the message and process it.

• Once the messages are processed, consumer will send an acknowledgement to the Kafka broker.

• Once Kafka receives an acknowledgement, it changes the offset to the new value and updates it in the Zookeeper. Since offsets are maintained in the Zookeeper, the consumer can read next message correctly even during server outrages.

• This above flow will repeat until the consumer stops the request.

• Consumer has the option to rewind/skip to the desired offset of a topic at any time and read all the subsequent messages.

Workflow of Queue Messaging / Consumer Group

In a queue messaging system instead of a single consumer, a group of consumers having the same Group ID will subscribe to a topic. In simple terms, consumers subscribing to a topic with same Group ID are considered as a single group and the messages are shared among them. Let us check the actual workflow of this system.

• Producers send message to a topic in a regular interval.

• Kafka stores all messages in the partitions configured for that particular topic similar to the earlier scenario.

• A single consumer subscribes to a specific topic, assume Topic-01 with Group ID as Group-1.

• Kafka interacts with the consumer in the same way as Pub-Sub Messaging until new consumer subscribes the same topic, Topic-01 with the same Group ID as Group-1.

• Once the new consumer arrives, Kafka switches its operation to share mode and shares the data between the two consumers. This sharing will go on until the number of con-sumers reach the number of partition configured for that particular topic.

• Once the number of consumer exceeds the number of partitions, the new consumer will not receive any further message until any one of the existing consumer unsubscribes. This scenario arises because each consumer in Kafka will be assigned a minimum of one partition and once all the partitions are assigned to the existing consumers, the new consumers will have to wait.

• This feature is also called as Consumer Group. In the same way, Kafka will provide the best of both the systems in a very simple and efficient manner.

Role of ZooKeeper

A critical dependency of Apache Kafka is Apache Zookeeper, which is a distributed configuration and synchronization service. Zookeeper serves as the coordination interface between the Kafka brokers and consumers. The Kafka servers share information via a Zookeeper cluster. Kafka stores basic metadata in Zookeeper such as information about topics, brokers, consumer offsets (queue readers) and so on.

Since all the critical information is stored in the Zookeeper and it normally replicates this data across its ensemble, failure of Kafka broker / Zookeeper does not affect the state of the Kafka cluster. Kafka will restore the state, once the Zookeeper restarts. This gives zero downtime for Kafka. The leader election between the Kafka broker is also done by using Zookeeper in the event of leader failure.

To learn more on Zookeeper, please refer zookeeper

Let us continue further on how to install Java, ZooKeeper, and Kafka on your machine in the next chapter.

Following are the steps for installing Java on your machine.

Step 1 - Verifying Java Installation

Hopefully you have already installed java on your machine right now, so you just verify it using the following command.

$ java -version

If java is successfully installed on your machine, you could see the version of the installed Java.

Step 1.1 - Download JDK

If Java is not downloaded, please download the latest version of JDK by visiting the following link and download latest version.

http://www.oracle.com/technetwork/java/javase/downloads/index.html

Now the latest version is JDK 8u 60 and the file is “jdk-8u60-linux-x64.tar.gz”. Please download the file on your machine.

Step 1.2 - Extract Files

Generally, files being downloaded are stored in the downloads folder, verify it and extract the tar setup using the following commands.

$ cd /go/to/download/path
$ tar -zxf jdk-8u60-linux-x64.gz

Step 1.3 - Move to Opt Directory

To make java available to all users, move the extracted java content to usr/local/java/ folder.

$ su
password: (type password of root user)
$ mkdir /opt/jdk $ mv jdk-1.8.0_60 /opt/jdk/

Step 1.4 - Set path

To set path and JAVA_HOME variables, add the following commands to ~/.bashrc file.

export JAVA_HOME =/usr/jdk/jdk-1.8.0_60
export PATH=$PATH:$JAVA_HOME/bin

Now apply all the changes into current running system.

$ source ~/.bashrc

Step 1.5 - Java Alternatives

Use the following command to change Java Alternatives.

update-alternatives --install /usr/bin/java java /opt/jdk/jdk1.8.0_60/bin/java 100

Step 1.6 − Now verify java using verification command (java -version) explained in Step 1.

Step 2 - ZooKeeper Framework Installation

Step 2.1 - Download ZooKeeper

To install ZooKeeper framework on your machine, visit the following link and download the latest version of ZooKeeper.

http://zookeeper.apache.org/releases.html

As of now, latest version of ZooKeeper is 3.4.6 (ZooKeeper-3.4.6.tar.gz).

Step 2.2 - Extract tar file

Extract tar file using the following command

$ cd opt/
$ tar -zxf zookeeper-3.4.6.tar.gz $ cd zookeeper-3.4.6
$ mkdir data

Step 2.3 - Create Configuration File

Open Configuration File named conf/zoo.cfg using the command vi “conf/zoo.cfg” and all the following parameters to set as starting point.

$ vi conf/zoo.cfg
tickTime=2000
dataDir=/path/to/zookeeper/data
clientPort=2181
initLimit=5
syncLimit=2

Once the configuration file has been saved successfully and return to terminal again, you can start the zookeeper server.

Step 2.4 - Start ZooKeeper Server

$ bin/zkServer.sh start

After executing this command, you will get a response as shown below −

$ JMX enabled by default
$ Using config: /Users/../zookeeper-3.4.6/bin/../conf/zoo.cfg $ Starting zookeeper ... STARTED

Step 2.5 - Start CLI

$ bin/zkCli.sh

After typing the above command, you will be connected to the zookeeper server and will get the below response.

Connecting to localhost:2181
................
................
................
Welcome to ZooKeeper!
................
................
WATCHER::
WatchedEvent state:SyncConnected type: None path:null
[zk: localhost:2181(CONNECTED) 0]

Step 2.6 - Stop Zookeeper Server

After connecting the server and performing all the operations, you can stop the zookeeper server with the following command −

$ bin/zkServer.sh stop

Now you have successfully installed Java and ZooKeeper on your machine. Let us see the steps to install Apache Kafka.

Step 3 - Apache Kafka Installation

Let us continue with the following steps to install Kafka on your machine.

Step 3.1 - Download Kafka

To install Kafka on your machine, click on the below link −

https://www.apache.org/dyn/closer.cgi?path=/kafka/0.9.0.0/kafka_2.11-0.9.0.0.tgz

Now the latest version i.e., – kafka_2.11_0.9.0.0.tgz will be downloaded onto your machine.

Step 3.2 - Extract the tar file

Extract the tar file using the following command −

$ cd opt/ $ tar -zxf kafka_2.11.0.9.0.0 tar.gz
$ cd kafka_2.11.0.9.0.0

Now you have downloaded the latest version of Kafka on your machine.

Step 3.3 - Start Server

You can start the server by giving the following command −

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

After the server starts, you would see the below response on your screen −

$ bin/kafka-server-start.sh config/server.properties
[2016-01-02 15:37:30,410] INFO KafkaConfig values:
request.timeout.ms = 30000
log.roll.hours = 168
inter.broker.protocol.version = 0.9.0.X
log.preallocate = false
security.inter.broker.protocol = PLAINTEXT
…………………………………………….
…………………………………………….

Step 4 - Stop the Server

After performing all the operations, you can stop the server using the following command −

$ bin/kafka-server-stop.sh config/server.properties

Now that we have already discussed the Kafka installation, we can learn how to perform basic operations on Kafka in the next chapter.

First let us start implementing single node-single broker configuration and we will then migrate our setup to single node-multiple brokers configuration.

Hopefully you would have installed Java, ZooKeeper and Kafka on your machine by now. Before moving to the Kafka Cluster Setup, first you would need to start your ZooKeeper because Kafka Cluster uses ZooKeeper.

Start ZooKeeper

Open a new terminal and type the following command −

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

To start Kafka Broker, type the following command −

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

After starting Kafka Broker, type the command jps on ZooKeeper terminal and you would see the following response −

821 QuorumPeerMain
928 Kafka
931 Jps

Now you could see two daemons running on the terminal where QuorumPeerMain is ZooKeeper daemon and another one is Kafka daemon.

Single Node-Single Broker Configuration

In this configuration you have a single ZooKeeper and broker id instance. Following are the steps to configure it −

Creating a Kafka Topic − Kafka provides a command line utility named kafka-topics.sh to create topics on the server. Open new terminal and type the below example.

Syntax

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 
--partitions 1 --topic topic-name

Example

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1   
--partitions 1 --topic Hello-Kafka

We just created a topic named Hello-Kafka with a single partition and one replica factor. The above created output will be similar to the following output −

Output − Created topic Hello-Kafka

Once the topic has been created, you can get the notification in Kafka broker terminal window and the log for the created topic specified in “/tmp/kafka-logs/“ in the config/server.properties file.

List of Topics

To get a list of topics in Kafka server, you can use the following command −

Syntax

bin/kafka-topics.sh --list --zookeeper localhost:2181

Output

Hello-Kafka

Since we have created a topic, it will list out Hello-Kafka only. Suppose, if you create more than one topics, you will get the topic names in the output.

Start Producer to Send Messages

Syntax

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-name

From the above syntax, two main parameters are required for the producer command line client −

Broker-list − The list of brokers that we want to send the messages to. In this case we only have one broker. The Config/server.properties file contains broker port id, since we know our broker is listening on port 9092, so you can specify it directly.

Topic name − Here is an example for the topic name.

Example

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka

The producer will wait on input from stdin and publishes to the Kafka cluster. By default, every new line is published as a new message then the default producer properties are specified in config/producer.properties file. Now you can type a few lines of messages in the terminal as shown below.

Output

$ bin/kafka-console-producer.sh --broker-list localhost:9092 
--topic Hello-Kafka[2016-01-16 13:50:45,931] 
WARN property topic is not valid (kafka.utils.Verifia-bleProperties)
Hello
My first message

My second message

Start Consumer to Receive Messages

Similar to producer, the default consumer properties are specified in config/consumer.proper-ties file. Open a new terminal and type the below syntax for consuming messages.

Syntax

bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic topic-name 
--from-beginning

Example

bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-Kafka 
--from-beginning

Output

Hello
My first message
My second message

Finally, you are able to enter messages from the producer’s terminal and see them appearing in the consumer’s terminal. As of now, you have a very good understanding on the single node cluster with a single broker. Let us now move on to the multiple brokers configuration.

Single Node-Multiple Brokers Configuration

Before moving on to the multiple brokers cluster setup, first start your ZooKeeper server.

Create Multiple Kafka Brokers − We have one Kafka broker instance already in con-fig/server.properties. Now we need multiple broker instances, so copy the existing server.prop-erties file into two new config files and rename it as server-one.properties and server-two.prop-erties. Then edit both new files and assign the following changes −

config/server-one.properties

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1
# The port the socket server listens on
port=9093
# A comma seperated list of directories under which to store log files
log.dirs=/tmp/kafka-logs-1

config/server-two.properties

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=2
# The port the socket server listens on
port=9094
# A comma seperated list of directories under which to store log files
log.dirs=/tmp/kafka-logs-2

Start Multiple Brokers− After all the changes have been made on three servers then open three new terminals to start each broker one by one.

Broker1
bin/kafka-server-start.sh config/server.properties
Broker2
bin/kafka-server-start.sh config/server-one.properties
Broker3
bin/kafka-server-start.sh config/server-two.properties

Now we have three different brokers running on the machine. Try it by yourself to check all the daemons by typing jps on the ZooKeeper terminal, then you would see the response.

Creating a Topic

Let us assign the replication factor value as three for this topic because we have three different brokers running. If you have two brokers, then the assigned replica value will be two.

Syntax

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 
-partitions 1 --topic topic-name

Example

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 
-partitions 1 --topic Multibrokerapplication

Output

created topic “Multibrokerapplication”

The Describe command is used to check which broker is listening on the current created topic as shown below −

bin/kafka-topics.sh --describe --zookeeper localhost:2181 
--topic Multibrokerappli-cation

Output

bin/kafka-topics.sh --describe --zookeeper localhost:2181 
--topic Multibrokerappli-cation

Topic:Multibrokerapplication    PartitionCount:1 
ReplicationFactor:3 Configs:
   
Topic:Multibrokerapplication Partition:0 Leader:0 
Replicas:0,2,1 Isr:0,2,1

From the above output, we can conclude that first line gives a summary of all the partitions, showing topic name, partition count and the replication factor that we have chosen already. In the second line, each node will be the leader for a randomly selected portion of the partitions.

In our case, we see that our first broker (with broker.id 0) is the leader. Then Replicas:0,2,1 means that all the brokers replicate the topic finally Isr is the set of in-sync replicas. Well, this is the subset of replicas that are currently alive and caught up by the leader.

Start Producer to Send Messages

This procedure remains the same as in the single broker setup.

Example

bin/kafka-console-producer.sh --broker-list localhost:9092 
--topic Multibrokerapplication

Output

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Multibrokerapplication
[2016-01-20 19:27:21,045] WARN Property topic is not valid (kafka.utils.Verifia-bleProperties)
This is single node-multi broker demo
This is the second message

Start Consumer to Receive Messages

This procedure remains the same as shown in the single broker setup.

Example

bin/kafka-console-consumer.sh --zookeeper localhost:2181 
—topic Multibrokerapplica-tion --from-beginning

Output

bin/kafka-console-consumer.sh --zookeeper localhost:2181 
—topic Multibrokerapplica-tion —from-beginning
This is single node-multi broker demo
This is the second message

Basic Topic Operations

In this chapter we will discuss the various basic topic operations.

Modifying a Topic

As you have already understood how to create a topic in Kafka Cluster. Now let us modify a created topic using the following command

Syntax

bin/kafka-topics.sh —zookeeper localhost:2181 --alter --topic topic_name 
--parti-tions count

Example

We have already created a topic “Hello-Kafka” with single partition count and one replica factor. 
Now using “alter” command we have changed the partition count.
bin/kafka-topics.sh --zookeeper localhost:2181 
--alter --topic Hello-kafka --parti-tions 2

Output

WARNING: If partitions are increased for a topic that has a key, 
the partition logic or ordering of the messages will be affected
Adding partitions succeeded!

Deleting a Topic

To delete a topic, you can use the following syntax.

Syntax

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic topic_name

Example

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic Hello-kafka

Output

> Topic Hello-kafka marked for deletion

Note −This will have no impact if delete.topic.enable is not set to true

Let us create an application for publishing and consuming messages using a Java client. Kafka producer client consists of the following API’s.

KafkaProducer API

Let us understand the most important set of Kafka producer API in this section. The central part of the KafkaProducer API is KafkaProducer class. The KafkaProducer class provides an option to connect a Kafka broker in its constructor with the following methods.

• KafkaProducer class provides send method to send messages asynchronously to a topic. The signature of send() is as follows

producer.send(new ProducerRecord<byte[],byte[]>(topic, 
partition, key1, value1) , callback);

• ProducerRecord − The producer manages a buffer of records waiting to be sent.

• Callback − A user-supplied callback to execute when the record has been acknowl-edged by the server (null indicates no callback).

• KafkaProducer class provides a flush method to ensure all previously sent messages have been actually completed. Syntax of the flush method is as follows −

public void flush()

• KafkaProducer class provides partitionFor method, which helps in getting the partition metadata for a given topic. This can be used for custom partitioning. The signature of this method is as follows −

public Map metrics()

It returns the map of internal metrics maintained by the producer.

• public void close() − KafkaProducer class provides close method blocks until all previously sent requests are completed.

Producer API

The central part of the Producer API is Producer class. Producer class provides an option to connect Kafka broker in its constructor by the following methods.

The Producer Class

The producer class provides send method to send messages to either single or multiple topics using the following signatures.

public void send(KeyedMessaget<k,v> message) 
- sends the data to a single topic,par-titioned by key using either sync or async producer.
public void send(List<KeyedMessage<k,v>>messages)
- sends data to multiple topics.
Properties prop = new Properties();
prop.put(producer.type,”async”)
ProducerConfig config = new ProducerConfig(prop);

There are two types of producers – Sync and Async.

The same API configuration applies to Sync producer as well. The difference between them is a sync producer sends messages directly, but sends messages in background. Async producer is preferred when you want a higher throughput. In the previous releases like 0.8, an async producer does not have a callback for send() to register error handlers. This is available only in the current release of 0.9.

public void close()

Producer class provides close method to close the producer pool connections to all Kafka bro-kers.

Configuration Settings

The Producer API’s main configuration settings are listed in the following table for better under-standing −

S.No	Configuration Settings and Description
1	client.id identifies producer application
2	producer.type either sync or async
3	acks The acks config controls the criteria under producer requests are con-sidered complete.
4	retries If producer request fails, then automatically retry with specific value.
5	bootstrap.servers bootstrapping list of brokers.
6	linger.ms if you want to reduce the number of requests you can set linger.ms to something greater than some value.
7	key.serializer Key for the serializer interface.
8	value.serializer value for the serializer interface.
9	batch.size Buffer size.
10	buffer.memory controls the total amount of memory available to the producer for buff-ering.

ProducerRecord API

ProducerRecord is a key/value pair that is sent to Kafka cluster.ProducerRecord class constructor for creating a record with partition, key and value pairs using the following signature.

public ProducerRecord (string topic, int partition, k key, v value)

• Topic − user defined topic name that will appended to record.

• Partition − partition count

• Key − The key that will be included in the record.

• Value − Record contents

public ProducerRecord (string topic, k key, v value)

ProducerRecord class constructor is used to create a record with key, value pairs and without partition.

• Topic − Create a topic to assign record.

• Key − key for the record.

• Value − record contents.

public ProducerRecord (string topic, v value)

ProducerRecord class creates a record without partition and key.

• Topic − create a topic.

• Value − record contents.

The ProducerRecord class methods are listed in the following table −

S.No	Class Methods and Description
1	public string topic() Topic will append to the record.
2	public K key() Key that will be included in the record. If no such key, null will be re-turned here.
3	public V value() Record contents.
4	partition() Partition count for the record

SimpleProducer application

Before creating the application, first start ZooKeeper and Kafka broker then create your own topic in Kafka broker using create topic command. After that create a java class named Sim-pleProducer.java and type in the following coding.

//import util.properties packages
import java.util.Properties;

//import simple producer packages
import org.apache.kafka.clients.producer.Producer;

//import KafkaProducer packages
import org.apache.kafka.clients.producer.KafkaProducer;

//import ProducerRecord packages
import org.apache.kafka.clients.producer.ProducerRecord;

//Create java class named “SimpleProducer”
public class SimpleProducer {
   
   public static void main(String[] args) throws Exception{
      
      // Check arguments length value
      if(args.length == 0){
         System.out.println("Enter topic name”);
         return;
      }
      
      //Assign topicName to string variable
      String topicName = args[0].toString();
      
      // create instance for properties to access producer configs   
      Properties props = new Properties();
      
      //Assign localhost id
      props.put("bootstrap.servers", “localhost:9092");
      
      //Set acknowledgements for producer requests.      
      props.put("acks", “all");
      
      //If the request fails, the producer can automatically retry,
      props.put("retries", 0);
      
      //Specify buffer size in config
      props.put("batch.size", 16384);
      
      //Reduce the no of requests less than 0   
      props.put("linger.ms", 1);
      
      //The buffer.memory controls the total amount of memory available to the producer for buffering.   
      props.put("buffer.memory", 33554432);
      
      props.put("key.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
         
      props.put("value.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
      
      Producer<String, String> producer = new KafkaProducer
         <String, String>(props);
            
      for(int i = 0; i < 10; i++)
         producer.send(new ProducerRecord<String, String>(topicName, 
            Integer.toString(i), Integer.toString(i)));
               System.out.println(“Message sent successfully”);
               producer.close();
   }
}

Compilation − The application can be compiled using the following command.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution − The application can be executed using the following command.

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleProducer <topic-name>

Output

Message sent successfully
To check the above output open new terminal and type Consumer CLI command to receive messages.
>> bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic <topic-name> —from-beginning
1
2
3
4
5
6
7
8
9
10

Simple Consumer Example

As of now we have created a producer to send messages to Kafka cluster. Now let us create a consumer to consume messages form the Kafka cluster. KafkaConsumer API is used to consume messages from the Kafka cluster. KafkaConsumer class constructor is defined below.

public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)

configs − Return a map of consumer configs.

KafkaConsumer class has the following significant methods that are listed in the table below.

S.No	Method and Description
1	public java.util.Set<TopicPar-tition> assignment() Get the set of partitions currently assigned by the con-sumer.
2	public string subscription() Subscribe to the given list of topics to get dynamically as-signed partitions.
3	public void sub-scribe(java.util.List<java.lang.String> topics, ConsumerRe-balanceListener listener) Subscribe to the given list of topics to get dynamically as-signed partitions.
4	public void unsubscribe() Unsubscribe the topics from the given list of partitions.
5	public void sub-scribe(java.util.List<java.lang.String> topics) Subscribe to the given list of topics to get dynamically as-signed partitions. If the given list of topics is empty, it is treated the same as unsubscribe().
6	public void sub-scribe(java.util.regex.Pattern pattern, ConsumerRebalanceLis-tener listener) The argument pattern refers to the subscribing pattern in the format of regular expression and the listener argument gets notifications from the subscribing pattern.
7	public void as-sign(java.util.List<TopicParti-tion> partitions) Manually assign a list of partitions to the customer.
8	poll() Fetch data for the topics or partitions specified using one of the subscribe/assign APIs. This will return error, if the topics are not subscribed before the polling for data.
9	public void commitSync() Commit offsets returned on the last poll() for all the sub-scribed list of topics and partitions. The same operation is applied to commitAsyn().
10	public void seek(TopicPartition partition, long offset) Fetch the current offset value that consumer will use on the next poll() method.
11	public void resume() Resume the paused partitions.
12	public void wakeup() Wakeup the consumer.

ConsumerRecord API

The ConsumerRecord API is used to receive records from the Kafka cluster. This API consists of a topic name, partition number, from which the record is being received and an offset that points to the record in a Kafka partition. ConsumerRecord class is used to create a consumer record with specific topic name, partition count and <key, value> pairs. It has the following signature.

public ConsumerRecord(string topic,int partition, long offset,K key, V value)

• Topic − The topic name for consumer record received from the Kafka cluster.

• Partition − Partition for the topic.

• Key − The key of the record, if no key exists null will be returned.

• Value − Record contents.

ConsumerRecords API

ConsumerRecords API acts as a container for ConsumerRecord. This API is used to keep the list of ConsumerRecord per partition for a particular topic. Its Constructor is defined below.

public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
<Consumer-Record>K,V>>> records)

• TopicPartition − Return a map of partition for a particular topic.

• Records − Return list of ConsumerRecord.

ConsumerRecords class has the following methods defined.

S.No	Methods and Description
1	public int count() The number of records for all the topics.
2	public Set partitions() The set of partitions with data in this record set (if no data was returned then the set is empty).
3	public Iterator iterator() Iterator enables you to cycle through a collection, obtaining or re-moving elements.
4	public List records() Get list of records for the given partition.

Paramètres de configuration

Les paramètres de configuration des principaux paramètres de configuration de l'API client client sont répertoriés ci-dessous:

S. Non	Paramètres et description
1	bootstrap.servers Liste de démarrage des courtiers.
2	group.id Affecte un consommateur individuel à un groupe.
3	enable.auto.commit Activez la validation automatique pour les décalages si la valeur est true, sinon non validée.
4	auto.commit.interval.ms Renvoie la fréquence à laquelle les décalages consommés mis à jour sont écrits dans ZooKeeper.
5	session.timeout.ms Indique combien de millisecondes Kafka attendra que le ZooKeeper réponde à une demande (lecture ou écriture) avant d'abandonner et de continuer à consommer des messages.

Application SimpleConsumer

Les étapes de candidature du producteur restent ici les mêmes. Tout d'abord, démarrez votre courtier ZooKeeper et Kafka. Créez ensuite une application SimpleConsumer avec la classe java nommée SimpleCon-sumer.java et saisissez le code suivant.

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class SimpleConsumer {
   public static void main(String[] args) throws Exception {
      if(args.length == 0){
         System.out.println("Enter topic name");
         return;
      }
      //Kafka consumer configuration settings
      String topicName = args[0].toString();
      Properties props = new Properties();
      
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", "test");
      props.put("enable.auto.commit", "true");
      props.put("auto.commit.interval.ms", "1000");
      props.put("session.timeout.ms", "30000");
      props.put("key.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer
         <String, String>(props);
      
      //Kafka Consumer subscribes list of topics here.
      consumer.subscribe(Arrays.asList(topicName))
      
      //print the topic name
      System.out.println("Subscribed to topic " + topicName);
      int i = 0;
      
      while (true) {
         ConsumerRecords<String, String> records = con-sumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
         
         // print the offset,key and value for the consumer records.
         System.out.printf("offset = %d, key = %s, value = %s\n", 
            record.offset(), record.key(), record.value());
      }
   }
}

Compilation - L'application peut être compilée à l'aide de la commande suivante.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution − L'application peut être exécutée à l'aide de la commande suivante

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleConsumer <topic-name>

Input- Ouvrez la CLI du producteur et envoyez des messages à la rubrique. Vous pouvez mettre l'entrée smple comme «Bonjour consommateur».

Output - Voici la sortie.

Subscribed to topic Hello-Kafka
offset = 3, key = null, value = Hello Consumer

Le groupe de consommateurs est une consommation multi-thread ou multi-machine à partir de sujets Kafka.

Groupe de consommateurs

• Les consommateurs peuvent rejoindre un groupe en utilisant le même group.id.

• Le parallélisme maximum d'un groupe est que le nombre de consommateurs dans le groupe ← no de partitions.

• Kafka attribue les partitions d'un sujet au consommateur d'un groupe, de sorte que chaque partition soit consommée par exactement un consommateur du groupe.

• Kafka garantit qu'un message n'est lu que par un seul consommateur du groupe.

• Les consommateurs peuvent voir le message dans l'ordre dans lequel ils ont été stockés dans le journal.

Rééquilibrage d'un consommateur

L'ajout de plus de processus / threads entraînera le rééquilibrage de Kafka. Si un consommateur ou un courtier ne parvient pas à envoyer des pulsations à ZooKeeper, il peut être reconfiguré via le cluster Kafka. Au cours de ce rééquilibrage, Kafka attribuera des partitions disponibles aux threads disponibles, déplaçant éventuellement une partition vers un autre processus.

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class ConsumerGroup {
   public static void main(String[] args) throws Exception {
      if(args.length < 2){
         System.out.println("Usage: consumer <topic> <groupname>");
         return;
      }
      
      String topic = args[0].toString();
      String group = args[1].toString();
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", group);
      props.put("enable.auto.commit", "true");
      props.put("auto.commit.interval.ms", "1000");
      props.put("session.timeout.ms", "30000");
      props.put("key.deserializer",          
         "org.apache.kafka.common.serialization.ByteArraySerializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
      
      consumer.subscribe(Arrays.asList(topic));
      System.out.println("Subscribed to topic " + topic);
      int i = 0;
         
      while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
               System.out.printf("offset = %d, key = %s, value = %s\n", 
               record.offset(), record.key(), record.value());
      }     
   }  
}

Compilation

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*" ConsumerGroup.java

Exécution

>>java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*":. 
ConsumerGroup <topic-name> my-group
>>java -cp "/home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*":. 
ConsumerGroup <topic-name> my-group

Ici, nous avons créé un exemple de nom de groupe en tant que mon-groupe avec deux consommateurs. De même, vous pouvez créer votre groupe et le nombre de consommateurs dans le groupe.

Contribution

Ouvrez la CLI du producteur et envoyez des messages comme -

Test consumer group 01
Test consumer group 02

Résultat du premier processus

Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 01

Résultat du deuxième processus

Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 02

J'espère que vous auriez compris SimpleConsumer et ConsumeGroup en utilisant la démo du client Java. Vous avez maintenant une idée sur la façon d'envoyer et de recevoir des messages à l'aide d'un client Java. Continuons l'intégration de Kafka avec les technologies Big Data dans le chapitre suivant.

Dans ce chapitre, nous allons apprendre à intégrer Kafka à Apache Storm.

À propos de Storm

Storm a été créé à l'origine par Nathan Marz et l'équipe de BackType. En peu de temps, Apache Storm est devenu un standard pour le système de traitement en temps réel distribué qui vous permet de traiter un énorme volume de données. Storm est très rapide et un benchmark l'a cadencé à plus d'un million de tuples traités par seconde et par nœud. Apache Storm s'exécute en continu, consomme les données des sources configurées (Spouts) et transmet les données dans le pipeline de traitement (Bolts). Combinés, les becs et les boulons forment une topologie.

Intégration avec Storm

Kafka et Storm se complètent naturellement, et leur puissante coopération permet des analyses de streaming en temps réel pour des données volumineuses en évolution rapide. L'intégration de Kafka et Storm permet aux développeurs d'ingérer et de publier plus facilement des flux de données à partir de topologies Storm.

Flux conceptuel

Un bec verseur est une source de flux. Par exemple, un spout peut lire des tuples d'un sujet Kafka et les émettre sous forme de flux. Un boulon consomme des flux d'entrée, traite et émet éventuellement de nouveaux flux. Les Bolts peuvent tout faire: exécuter des fonctions, filtrer des tuples, effectuer des agrégations de streaming, des jointures en streaming, parler à des bases de données, etc. Chaque nœud d'une topologie Storm s'exécute en parallèle. Une topologie s'exécute indéfiniment jusqu'à ce que vous la terminiez. Storm réaffectera automatiquement toutes les tâches ayant échoué. De plus, Storm garantit qu'il n'y aura aucune perte de données, même si les machines tombent en panne et que les messages sont supprimés.

Passons en revue les API d'intégration Kafka-Storm en détail. Il existe trois classes principales pour intégrer Kafka à Storm. Ils sont les suivants -

BrokerHosts - ZkHosts et hôtes statiques

BrokerHosts est une interface et ZkHosts et StaticHosts sont ses deux principales implémentations. ZkHosts est utilisé pour suivre dynamiquement les courtiers Kafka en conservant les détails dans ZooKeeper, tandis que StaticHosts est utilisé pour définir manuellement / statiquement les courtiers Kafka et ses détails. ZkHosts est le moyen simple et rapide d'accéder au courtier Kafka.

La signature de ZkHosts est la suivante -

public ZkHosts(String brokerZkStr, String brokerZkPath)
public ZkHosts(String brokerZkStr)

Où brokerZkStr est l'hôte ZooKeeper et brokerZkPath est le chemin ZooKeeper pour gérer les détails du courtier Kafka.

API KafkaConfig

Cette API est utilisée pour définir les paramètres de configuration du cluster Kafka. La signature de Kafka Con-fig est définie comme suit

public KafkaConfig(BrokerHosts hosts, string topic)

Hosts - Les BrokerHosts peuvent être des ZkHosts / StaticHosts.

Topic - nom du sujet.

API SpoutConfig

Spoutconfig est une extension de KafkaConfig qui prend en charge des informations supplémentaires sur ZooKeeper.

public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)

• Hosts - Les BrokerHosts peuvent être n'importe quelle implémentation de l'interface BrokerHosts

• Topic - nom du sujet.

• zkRoot - Chemin racine de ZooKeeper.

• id −Le bec stocke l'état des décalages consommés dans Zookeeper. L'identifiant doit identifier de manière unique votre bec.

SchemeAsMultiScheme

SchemeAsMultiScheme est une interface qui dicte comment le ByteBuffer consommé par Kafka est transformé en un tuple de tempête. Il est dérivé de MultiScheme et accepte l'implémentation de la classe Scheme. Il existe de nombreuses implémentations de la classe Scheme et l'une de ces implémentations est StringScheme, qui analyse l'octet comme une simple chaîne. Il contrôle également la dénomination de votre champ de sortie. La signature est définie comme suit.

public SchemeAsMultiScheme(Scheme scheme)

• Scheme - tampon d'octets consommé par kafka.

API KafkaSpout

KafkaSpout est notre implémentation de bec, qui s'intégrera à Storm. Il récupère les messages du sujet kafka et les émet dans l'écosystème Storm sous forme de tuples. KafkaSpout obtient ses détails de configuration de SpoutConfig.

Vous trouverez ci-dessous un exemple de code pour créer un simple bec verseur Kafka.

// ZooKeeper connection string
BrokerHosts hosts = new ZkHosts(zkConnString);

//Creating SpoutConfig Object
SpoutConfig spoutConfig = new SpoutConfig(hosts, 
   topicName, "/" + topicName UUID.randomUUID().toString());

//convert the ByteBuffer to String.
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

//Assign SpoutConfig to KafkaSpout.
KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

Création de boulons

Bolt est un composant qui prend des tuples en entrée, traite le tuple et produit de nouveaux tuples en sortie. Bolts implémentera l'interface IRichBolt. Dans ce programme, deux classes de boulons WordSplitter-Bolt et WordCounterBolt sont utilisées pour effectuer les opérations.

L'interface IRichBolt a les méthodes suivantes -

• Prepare- Fournit au boulon un environnement à exécuter. Les exécuteurs exécuteront cette méthode pour initialiser le bec.

• Execute - Traitez un seul tuple d'entrée.

• Cleanup - Appelé quand un verrou va s'arrêter.

• declareOutputFields - Déclare le schéma de sortie du tuple.

Créons SplitBolt.java, qui implémente la logique pour diviser une phrase en mots et CountBolt.java, qui implémente la logique pour séparer les mots uniques et compter son occurrence.

SplitBolt.java

import java.util.Map;

import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import backtype.storm.task.OutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.task.TopologyContext;

public class SplitBolt implements IRichBolt {
   private OutputCollector collector;
   
   @Override
   public void prepare(Map stormConf, TopologyContext context,
      OutputCollector collector) {
      this.collector = collector;
   }
   
   @Override
   public void execute(Tuple input) {
      String sentence = input.getString(0);
      String[] words = sentence.split(" ");
      
      for(String word: words) {
         word = word.trim();
         
         if(!word.isEmpty()) {
            word = word.toLowerCase();
            collector.emit(new Values(word));
         }
         
      }

      collector.ack(input);
   }
   
   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("word"));
   }

   @Override
   public void cleanup() {}
   
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
   
}

CountBolt.java

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

import backtype.storm.tuple.Tuple;
import backtype.storm.task.OutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.task.TopologyContext;

public class CountBolt implements IRichBolt{
   Map<String, Integer> counters;
   private OutputCollector collector;
   
   @Override
   public void prepare(Map stormConf, TopologyContext context,
   OutputCollector collector) {
      this.counters = new HashMap<String, Integer>();
      this.collector = collector;
   }

   @Override
   public void execute(Tuple input) {
      String str = input.getString(0);
      
      if(!counters.containsKey(str)){
         counters.put(str, 1);
      }else {
         Integer c = counters.get(str) +1;
         counters.put(str, c);
      }
   
      collector.ack(input);
   }

   @Override
   public void cleanup() {
      for(Map.Entry<String, Integer> entry:counters.entrySet()){
         System.out.println(entry.getKey()+" : " + entry.getValue());
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
   
   }

   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
}

Soumission à la topologie

La topologie Storm est essentiellement une structure Thrift. La classe TopologyBuilder fournit des méthodes simples et faciles pour créer des topologies complexes. La classe TopologyBuilder a des méthodes pour définir spout (setSpout) et pour définir bolt (setBolt). Enfin, TopologyBuilder a createTopology pour créer des to-pology. shuffleGrouping et champs Les méthodes de groupage aident à définir le regroupement de flux pour le bec et les boulons.

Local Cluster- Pour des fins de développement, nous pouvons créer un cluster local en utilisant LocalCluster objet, puis soumettre la topologie à l' aide submitTopology méthode de LocalCluster classe.

KafkaStormSample.java

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import backtype.storm.spout.SchemeAsMultiScheme;
import storm.kafka.trident.GlobalPartitionInformation;
import storm.kafka.ZkHosts;
import storm.kafka.Broker;
import storm.kafka.StaticHosts;
import storm.kafka.BrokerHosts;
import storm.kafka.SpoutConfig;
import storm.kafka.KafkaConfig;
import storm.kafka.KafkaSpout;
import storm.kafka.StringScheme;

public class KafkaStormSample {
   public static void main(String[] args) throws Exception{
      Config config = new Config();
      config.setDebug(true);
      config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
      String zkConnString = "localhost:2181";
      String topic = "my-first-topic";
      BrokerHosts hosts = new ZkHosts(zkConnString);
      
      SpoutConfig kafkaSpoutConfig = new SpoutConfig (hosts, topic, "/" + topic,    
         UUID.randomUUID().toString());
      kafkaSpoutConfig.bufferSizeBytes = 1024 * 1024 * 4;
      kafkaSpoutConfig.fetchSizeBytes = 1024 * 1024 * 4;
      kafkaSpoutConfig.forceFromStart = true;
      kafkaSpoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("kafka-spout", new KafkaSpout(kafkaSpoutCon-fig));
      builder.setBolt("word-spitter", new SplitBolt()).shuffleGroup-ing("kafka-spout");
      builder.setBolt("word-counter", new CountBolt()).shuffleGroup-ing("word-spitter");
         
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("KafkaStormSample", config, builder.create-Topology());

      Thread.sleep(10000);
      
      cluster.shutdown();
   }
}

Avant de déplacer la compilation, l'intégration Kakfa-Storm nécessite la bibliothèque java du client ZooKeeper du conservateur. La version 2.9.1 de Curator prend en charge la version 0.9.5 d'Apache Storm (que nous utilisons dans ce tutoriel). Téléchargez les fichiers jar spécifiés ci-dessous et placez-les dans le chemin de classe java.

• curator-client-2.9.1.jar
• curator-framework-2.9.1.jar

Après avoir inclus les fichiers de dépendance, compilez le programme à l'aide de la commande suivante,

javac -cp "/path/to/Kafka/apache-storm-0.9.5/lib/*" *.java

Exécution

Démarrez Kafka Producer CLI (expliqué dans le chapitre précédent), créez un nouveau sujet appelé my-first-topic et fournissez des exemples de messages comme indiqué ci-dessous -

hello
kafka
storm
spark
test message
another test message

Exécutez maintenant l'application à l'aide de la commande suivante -

java -cp “/path/to/Kafka/apache-storm-0.9.5/lib/*”:. KafkaStormSample

L'exemple de sortie de cette application est spécifié ci-dessous -

storm : 1
test : 2
spark : 1
another : 1
kafka : 1
hello : 1
message : 2

Dans ce chapitre, nous discuterons de la manière d'intégrer Apache Kafka avec l'API Spark Streaming.

À propos de Spark

L'API Spark Streaming permet un traitement de flux évolutif, à haut débit et tolérant aux pannes des flux de données en direct. Les données peuvent être ingérées à partir de nombreuses sources telles que Kafka, Flume, Twitter, etc., et peuvent être traitées à l'aide d'algorithmes complexes tels que des fonctions de haut niveau telles que mapper, réduire, joindre et fenêtre. Enfin, les données traitées peuvent être transférées vers des systèmes de fichiers, des bases de données et des tableaux de bord en direct. Les ensembles de données distribués résilients (RDD) sont une structure de données fondamentale de Spark. Il s'agit d'une collection d'objets distribués immuable. Chaque ensemble de données dans RDD est divisé en partitions logiques, qui peuvent être calculées sur différents nœuds du cluster.

Intégration avec Spark

Kafka est une plate-forme potentielle de messagerie et d'intégration pour le streaming Spark. Kafka sert de hub central pour les flux de données en temps réel et sont traités à l'aide d'algorithmes complexes dans Spark Streaming. Une fois les données traitées, Spark Streaming pourrait publier les résultats dans un autre sujet Kafka ou stocker dans HDFS, des bases de données ou des tableaux de bord. Le diagramme suivant illustre le flux conceptuel.

Passons maintenant en revue les API Kafka-Spark en détail.

API SparkConf

Il représente la configuration d'une application Spark. Utilisé pour définir divers paramètres Spark sous forme de paires clé-valeur.

La classe SparkConf a les méthodes suivantes -

• set(string key, string value) - définir la variable de configuration.

• remove(string key) - retirer la clé de la configuration.

• setAppName(string name) - définissez le nom de l'application pour votre application.

• get(string key) - obtenir la clé

API StreamingContext

C'est le principal point d'entrée pour la fonctionnalité Spark. Un SparkContext représente la connexion à un cluster Spark et peut être utilisé pour créer des RDD, des accumulateurs et des variables de diffusion sur le cluster. La signature est définie comme indiqué ci-dessous.

public StreamingContext(String master, String appName, Duration batchDuration, 
   String sparkHome, scala.collection.Seq<String> jars, 
   scala.collection.Map<String,String> environment)

• master - URL du cluster à laquelle se connecter (par exemple mesos: // hôte: port, spark: // hôte: port, local [4]).

• appName - un nom pour votre travail, à afficher sur l'interface utilisateur Web du cluster

• batchDuration - l'intervalle de temps auquel les données de streaming seront divisées en lots

public StreamingContext(SparkConf conf, Duration batchDuration)

Créez un StreamingContext en fournissant la configuration nécessaire pour un nouveau SparkContext.

• conf - Paramètres Spark

• batchDuration - l'intervalle de temps auquel les données de streaming seront divisées en lots

API KafkaUtils

L'API KafkaUtils est utilisée pour connecter le cluster Kafka au streaming Spark. Cette API a la signature createStream de la méthode significative définie comme ci-dessous.

public static ReceiverInputDStream<scala.Tuple2<String,String>> createStream(
   StreamingContext ssc, String zkQuorum, String groupId,
   scala.collection.immutable.Map<String,Object> topics, StorageLevel storageLevel)

La méthode illustrée ci-dessus est utilisée pour créer un flux d'entrée qui extrait les messages de Kafka Brokers.

• ssc - Objet StreamingContext.

• zkQuorum - Collège des gardiens de zoo.

• groupId - L'identifiant de groupe pour ce consommateur.

• topics - renvoyer une carte des sujets à consommer.

• storageLevel - Niveau de stockage à utiliser pour stocker les objets reçus.

L'API KafkaUtils a une autre méthode createDirectStream, qui est utilisée pour créer un flux d'entrée qui extrait directement les messages de Kafka Brokers sans utiliser de récepteur. Ce flux peut garantir que chaque message de Kafka est inclus dans les transformations exactement une fois.

L'exemple d'application est réalisé dans Scala. Pour compiler l'application, veuillez télécharger et installer sbt , scala build tool (similaire à maven). Le code principal de l'application est présenté ci-dessous.

import java.util.HashMap

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, Produc-erRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._

object KafkaWordCount {
   def main(args: Array[String]) {
      if (args.length < 4) {
         System.err.println("Usage: KafkaWordCount <zkQuorum><group> <topics> <numThreads>")
         System.exit(1)
      }

      val Array(zkQuorum, group, topics, numThreads) = args
      val sparkConf = new SparkConf().setAppName("KafkaWordCount")
      val ssc = new StreamingContext(sparkConf, Seconds(2))
      ssc.checkpoint("checkpoint")

      val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
      val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
      val words = lines.flatMap(_.split(" "))
      val wordCounts = words.map(x => (x, 1L))
         .reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(2), 2)
      wordCounts.print()

      ssc.start()
      ssc.awaitTermination()
   }
}

Créer un script

L'intégration spark-kafka dépend du jar d'intégration Spark, Spark Streaming et Spark Kafka. Créez un nouveau fichier build.sbt et spécifiez les détails de l'application et sa dépendance. Le sbt téléchargera le fichier jar nécessaire lors de la compilation et de l'emballage de l'application.

name := "Spark Kafka Project"
version := "1.0"
scalaVersion := "2.10.5"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming" % "1.6.0"
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka" % "1.6.0"

Compilation / Emballage

Exécutez la commande suivante pour compiler et empaqueter le fichier jar de l'application. Nous devons soumettre le fichier jar dans la console Spark pour exécuter l'application.

sbt package

Soumettre à Spark

Démarrez l'interface de ligne de commande de Kafka Producer (expliqué dans le chapitre précédent), créez un nouveau sujet appelé my-first-topic et fournissez des exemples de messages comme indiqué ci-dessous.

Another spark test message

Exécutez la commande suivante pour soumettre l'application à la console Spark.

/usr/local/spark/bin/spark-submit --packages org.apache.spark:spark-streaming
-kafka_2.10:1.6.0 --class "KafkaWordCount" --master local[4] target/scala-2.10/spark
-kafka-project_2.10-1.0.jar localhost:2181 <group name> <topic name> <number of threads>

L'exemple de sortie de cette application est illustré ci-dessous.

spark console messages ..
(Test,1)
(spark,1)
(another,1)
(message,1)
spark console message ..

Analysons une application en temps réel pour obtenir les derniers flux Twitter et ses hashtags. Auparavant, nous avons vu l'intégration de Storm et Spark avec Kafka. Dans les deux scénarios, nous avons créé un producteur Kafka (en utilisant cli) pour envoyer un message à l'écosystème Kafka. Ensuite, l'intégration tempête et étincelle lit les messages en utilisant le consommateur Kafka et l'injecte respectivement dans l'écosystème tempête et étincelle. Donc, pratiquement, nous devons créer un producteur Kafka, qui devrait -

• Lisez les flux Twitter à l'aide de «Twitter Streaming API»,
• Traiter les flux,
• Extraire les HashTags et
• Envoyez-le à Kafka.

Une fois les HashTags reçus par Kafka, l'intégration Storm / Spark reçoit les informations et les envoie à l'écosystème Storm / Spark.

API de streaming Twitter

L '«API Twitter Streaming» est accessible dans n'importe quel langage de programmation. «Twitter4j» est une bibliothèque Java non officielle open source, qui fournit un module basé sur Java pour accéder facilement à «Twitter Streaming API». Le «twitter4j» fournit un cadre basé sur l'auditeur pour accéder aux tweets. Pour accéder à «Twitter Streaming API», nous devons nous connecter au compte de développeur Twitter et obtenir les éléments suivantsOAuth détails d'authentification.

• Customerkey
• CustomerSecret
• AccessToken
• AccessTookenSecret

Une fois le compte développeur créé, téléchargez les fichiers jar «twitter4j» et placez-les dans le chemin de classe java.

Le codage complet du producteur Twitter Kafka (KafkaTwitterProducer.java) est répertorié ci-dessous -

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;

import twitter4j.*;
import twitter4j.conf.*;

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaTwitterProducer {
   public static void main(String[] args) throws Exception {
      LinkedBlockingQueue<Status> queue = new LinkedBlockingQueue<Sta-tus>(1000);
      
      if(args.length < 5){
         System.out.println(
            "Usage: KafkaTwitterProducer <twitter-consumer-key>
            <twitter-consumer-secret> <twitter-access-token>
            <twitter-access-token-secret>
            <topic-name> <twitter-search-keywords>");
         return;
      }
      
      String consumerKey = args[0].toString();
      String consumerSecret = args[1].toString();
      String accessToken = args[2].toString();
      String accessTokenSecret = args[3].toString();
      String topicName = args[4].toString();
      String[] arguments = args.clone();
      String[] keyWords = Arrays.copyOfRange(arguments, 5, arguments.length);

      ConfigurationBuilder cb = new ConfigurationBuilder();
      cb.setDebugEnabled(true)
         .setOAuthConsumerKey(consumerKey)
         .setOAuthConsumerSecret(consumerSecret)
         .setOAuthAccessToken(accessToken)
         .setOAuthAccessTokenSecret(accessTokenSecret);

      TwitterStream twitterStream = new TwitterStreamFactory(cb.build()).get-Instance();
      StatusListener listener = new StatusListener() {
        
         @Override
         public void onStatus(Status status) {      
            queue.offer(status);

            // System.out.println("@" + status.getUser().getScreenName() 
               + " - " + status.getText());
            // System.out.println("@" + status.getUser().getScreen-Name());

            /*for(URLEntity urle : status.getURLEntities()) {
               System.out.println(urle.getDisplayURL());
            }*/

            /*for(HashtagEntity hashtage : status.getHashtagEntities()) {
               System.out.println(hashtage.getText());
            }*/
         }
         
         @Override
         public void onDeletionNotice(StatusDeletionNotice statusDeletion-Notice) {
            // System.out.println("Got a status deletion notice id:" 
               + statusDeletionNotice.getStatusId());
         }
         
         @Override
         public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
            // System.out.println("Got track limitation notice:" + 
               num-berOfLimitedStatuses);
         }

         @Override
         public void onScrubGeo(long userId, long upToStatusId) {
            // System.out.println("Got scrub_geo event userId:" + userId + 
            "upToStatusId:" + upToStatusId);
         }      
         
         @Override
         public void onStallWarning(StallWarning warning) {
            // System.out.println("Got stall warning:" + warning);
         }
         
         @Override
         public void onException(Exception ex) {
            ex.printStackTrace();
         }
      };
      twitterStream.addListener(listener);
      
      FilterQuery query = new FilterQuery().track(keyWords);
      twitterStream.filter(query);

      Thread.sleep(5000);
      
      //Add Kafka producer config settings
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      props.put("acks", "all");
      props.put("retries", 0);
      props.put("batch.size", 16384);
      props.put("linger.ms", 1);
      props.put("buffer.memory", 33554432);
      
      props.put("key.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
      props.put("value.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
      
      Producer<String, String> producer = new KafkaProducer<String, String>(props);
      int i = 0;
      int j = 0;
      
      while(i < 10) {
         Status ret = queue.poll();
         
         if (ret == null) {
            Thread.sleep(100);
            i++;
         }else {
            for(HashtagEntity hashtage : ret.getHashtagEntities()) {
               System.out.println("Hashtag: " + hashtage.getText());
               producer.send(new ProducerRecord<String, String>(
                  top-icName, Integer.toString(j++), hashtage.getText()));
            }
         }
      }
      producer.close();
      Thread.sleep(5000);
      twitterStream.shutdown();
   }
}

Compilation

Compilez l'application à l'aide de la commande suivante -

javac -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”:. KafkaTwitterProducer.java

Exécution

Ouvrez deux consoles. Exécutez l'application compilée ci-dessus comme indiqué ci-dessous dans une console.

java -cp “/path/to/kafka/libs/*”:”/path/to/twitter4j/lib/*”:
. KafkaTwitterProducer <twitter-consumer-key>
<twitter-consumer-secret>
<twitter-access-token>
<twitter-ac-cess-token-secret>
my-first-topic food

Exécutez l'une des applications Spark / Storm expliquées dans le chapitre précédent dans une autre fenêtre. Le point principal à noter est que le sujet utilisé doit être le même dans les deux cas. Ici, nous avons utilisé «mon premier sujet» comme nom du sujet.

Production

La sortie de cette application dépendra des mots-clés et du flux actuel du twitter. Un exemple de sortie est spécifié ci-dessous (intégration de tempête).

. . .
food : 1
foodie : 2
burger : 1
. . .

Outil Kafka emballé sous «org.apache.kafka.tools. *. Les outils sont classés en outils système et outils de réplication.

Outils système

Les outils système peuvent être exécutés à partir de la ligne de commande à l'aide du script run class. La syntaxe est la suivante -

bin/kafka-run-class.sh package.class - - options

Certains des outils système sont mentionnés ci-dessous -

• Kafka Migration Tool - Cet outil permet de migrer un courtier d'une version à une autre.

• Mirror Maker - Cet outil est utilisé pour fournir la mise en miroir d'un cluster Kafka à un autre.

• Consumer Offset Checker - Cet outil affiche le groupe de consommateurs, le sujet, les partitions, le décalage, la taille du journal, le propriétaire pour l'ensemble spécifié de sujets et le groupe de consommateurs.

Outil de réplication

La réplication Kafka est un outil de conception de haut niveau. Le but de l'ajout d'un outil de réplication est une durabilité et une disponibilité plus élevées. Certains des outils de réplication sont mentionnés ci-dessous -

• Create Topic Tool - Cela crée une rubrique avec un nombre par défaut de partitions, un facteur de réplication et utilise le schéma par défaut de Kafka pour effectuer l'attribution de répliques.

• List Topic Tool- Cet outil répertorie les informations pour une liste de sujets donnée. Si aucune rubrique n'est fournie dans la ligne de commande, l'outil interroge Zookeeper pour obtenir toutes les rubriques et répertorie les informations les concernant. Les champs affichés par l'outil sont le nom du sujet, la partition, le leader, les répliques, isr.

• Add Partition Tool- Création d'un thème, le nombre de partitions pour le thème doit être spécifié. Plus tard, plus de partitions peuvent être nécessaires pour le sujet, lorsque le volume du sujet augmentera. Cet outil permet d'ajouter plus de partitions pour un sujet spécifique et permet également l'attribution manuelle de répliques des partitions ajoutées.

Kafka prend en charge plusieurs des meilleures applications industrielles actuelles. Nous fournirons un très bref aperçu de certaines des applications les plus notables de Kafka dans ce chapitre.

Twitter

Twitter est un service de réseau social en ligne qui fournit une plate-forme pour envoyer et recevoir des tweets d'utilisateurs. Les utilisateurs enregistrés peuvent lire et publier des tweets, mais les utilisateurs non enregistrés ne peuvent lire que des tweets. Twitter utilise Storm-Kafka dans le cadre de son infrastructure de traitement de flux.

LinkedIn

Apache Kafka est utilisé chez LinkedIn pour les données de flux d'activité et les mesures opérationnelles. Le système de messagerie Kafka aide LinkedIn avec divers produits tels que LinkedIn Newsfeed, LinkedIn Today pour la consommation de messages en ligne et en plus des systèmes d'analyse hors ligne comme Hadoop. La forte durabilité de Kafka est également l'un des facteurs clés en relation avec LinkedIn.

Netflix

Netflix est un fournisseur multinational américain de médias de streaming Internet à la demande. Netflix utilise Kafka pour la surveillance en temps réel et le traitement des événements.

Mozilla

Mozilla est une communauté de logiciels libres, créée en 1998 par des membres de Netscape. Kafka remplacera bientôt une partie du système de production actuel de Mozilla pour collecter les données de performances et d'utilisation du navigateur de l'utilisateur final pour des projets tels que la télémétrie, le pilote de test, etc.

Oracle

Oracle fournit une connectivité native à Kafka à partir de son produit Enterprise Service Bus appelé OSB (Oracle Service Bus), qui permet aux développeurs d'exploiter les capacités de médiation intégrées d'OSB pour implémenter des pipelines de données par étapes.