Apache Kafka - Integrasi Dengan Storm

Pada bab ini, kita akan belajar bagaimana mengintegrasikan Kafka dengan Apache Storm.

Tentang Storm

Storm awalnya dibuat oleh Nathan Marz dan tim di BackType. Dalam waktu singkat, Apache Storm menjadi standar untuk sistem pemrosesan waktu nyata terdistribusi yang memungkinkan Anda memproses data dalam jumlah besar. Storm sangat cepat dan patokan mencatatnya di lebih dari satu juta tupel yang diproses per detik per node. Apache Storm berjalan terus menerus, mengonsumsi data dari sumber yang dikonfigurasi (Spout) dan meneruskan data tersebut ke pipeline pemrosesan (Bolts). Gabungan, Cerat, dan Baut membuat Topologi.

Integrasi dengan Storm

Kafka dan Storm secara alami saling melengkapi, dan kerja sama mereka yang kuat memungkinkan analisis streaming waktu nyata untuk data besar yang bergerak cepat. Integrasi Kafka dan Storm mempermudah pengembang untuk menyerap dan memublikasikan aliran data dari topologi Storm.

Aliran konseptual

Cerat adalah sumber aliran. Misalnya, cerat dapat membaca tupel dari Topik Kafka dan memancarkannya sebagai aliran. Sebuah baut mengkonsumsi aliran input, memproses dan mungkin memancarkan aliran baru. Bolts dapat melakukan apa saja mulai dari menjalankan fungsi, memfilter tupel, melakukan agregasi streaming, streaming bergabung, berbicara dengan database, dan banyak lagi. Setiap node dalam topologi Storm dijalankan secara paralel. Topologi berjalan tanpa batas waktu hingga Anda menghentikannya. Storm secara otomatis akan menetapkan ulang tugas yang gagal. Selain itu, Storm menjamin bahwa tidak akan ada kehilangan data, bahkan jika mesin mati dan pesan dijatuhkan.

Mari kita telusuri API integrasi Kafka-Storm secara detail. Ada tiga kelas utama untuk mengintegrasikan Kafka dengan Storm. Mereka adalah sebagai berikut -

BrokerHosts - ZkHosts & StaticHosts

BrokerHosts adalah sebuah antarmuka dan ZkHosts dan StaticHosts adalah dua implementasi utamanya. ZkHosts digunakan untuk melacak broker Kafka secara dinamis dengan menjaga detail di ZooKeeper, sedangkan StaticHosts digunakan untuk mengatur broker Kafka secara manual / statis dan detailnya. ZkHosts adalah cara sederhana dan cepat untuk mengakses broker Kafka.

Tanda tangan ZkHosts adalah sebagai berikut -

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

Dimana brokerZkStr adalah host Zkeeper dan brokerZkPath adalah jalur ZooKeeper untuk menjaga detail broker Kafka.

API KafkaConfig

API ini digunakan untuk menentukan pengaturan konfigurasi untuk cluster Kafka. Tanda tangan Kafka Con-fig didefinisikan sebagai berikut

public KafkaConfig(BrokerHosts hosts, string topic)

Hosts - BrokerHosts dapat berupa ZkHosts / StaticHosts.

Topic - nama topik.

SpoutConfig API

Spoutconfig adalah perpanjangan dari KafkaConfig yang mendukung informasi tambahan ZooKeeper.

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

• Hosts - BrokerHosts dapat berupa implementasi antarmuka BrokerHosts apa pun

• Topic - nama topik.

• zkRoot - Jalur root ZooKeeper.

• id −Cerat menyimpan keadaan offset yang dikonsumsi di Zookeeper. Id harus mengidentifikasi cerat Anda secara unik.

SchemeAsMultiScheme

SchemeAsMultiScheme adalah antarmuka yang menentukan bagaimana ByteBuffer yang dikonsumsi dari Kafka diubah menjadi tupel badai. Ini berasal dari MultiScheme dan menerima implementasi kelas Skema. Ada banyak implementasi kelas Scheme dan salah satu implementasi tersebut adalah StringScheme, yang mem-parsing byte sebagai string sederhana. Ini juga mengontrol penamaan bidang output Anda. Tanda tangan tersebut didefinisikan sebagai berikut.

public SchemeAsMultiScheme(Scheme scheme)

• Scheme - buffer byte yang dikonsumsi dari kafka.

API KafkaSpout

KafkaSpout adalah implementasi cerat kami, yang akan berintegrasi dengan Storm. Ini mengambil pesan dari topik kafka dan memancarkannya ke ekosistem Storm sebagai tuple. KafkaSpout mendapatkan detail konfigurasi dari SpoutConfig.

Di bawah ini adalah contoh kode untuk membuat cerat Kafka sederhana.

// 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);

Pembuatan Baut

Bolt merupakan komponen yang mengambil tupel sebagai masukan, memproses tupel, dan menghasilkan tupel baru sebagai keluaran. Baut akan mengimplementasikan antarmuka IRichBolt. Dalam program ini, dua kelas baut WordSplitter-Bolt dan WordCounterBolt digunakan untuk menjalankan operasi.

Antarmuka IRichBolt memiliki metode berikut -

• Prepare- Menyediakan baut dengan lingkungan untuk dieksekusi. Pelaksana akan menjalankan metode ini untuk menginisialisasi cerat.

• Execute - Memproses satu tupel input.

• Cleanup - Dipanggil saat baut akan mati.

• declareOutputFields - Mendeklarasikan skema keluaran tupel.

Mari kita buat SplitBolt.java, yang mengimplementasikan logika untuk membagi kalimat menjadi kata-kata dan CountBolt.java, yang mengimplementasikan logika untuk memisahkan kata-kata unik dan menghitung kemunculannya.

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;
   }
}

Mengirimkan ke Topologi

Topologi Storm pada dasarnya adalah struktur Hemat. Kelas TopologyBuilder menyediakan metode sederhana dan mudah untuk membuat topologi yang kompleks. Kelas TopologyBuilder memiliki metode untuk menyetel cerat (setSpout) dan untuk menyetel baut (setBolt). Terakhir, TopologyBuilder memiliki createTopology untuk membuat to-pology. shuffleGrouping and fieldsGrouping method membantu mengatur pengelompokan aliran untuk cerat dan baut.

Local Cluster- Untuk tujuan pembangunan, kita dapat membuat cluster lokal menggunakan LocalCluster objek dan kemudian menyerahkan topologi menggunakan submitTopology metode LocalCluster kelas.

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();
   }
}

Sebelum memindahkan kompilasi, integrasi Kakfa-Storm membutuhkan java library kurator ZooKeeper client. Kurator versi 2.9.1 mendukung Apache Storm versi 0.9.5 (yang kami gunakan dalam tutorial ini). Unduh file jar yang ditentukan di bawah ini dan letakkan di jalur kelas java.

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

Setelah memasukkan file dependensi, kompilasi program menggunakan perintah berikut,

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

Eksekusi

Mulai Kafka Producer CLI (dijelaskan di bab sebelumnya), buat topik baru bernama my-first-topic dan berikan beberapa contoh pesan seperti yang ditunjukkan di bawah ini -

hello
kafka
storm
spark
test message
another test message

Sekarang jalankan aplikasi menggunakan perintah berikut -

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

Contoh keluaran dari aplikasi ini ditentukan di bawah -

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