Apache Kafka - การรวมเข้ากับ Storm

ในบทนี้เราจะเรียนรู้วิธีรวม Kafka กับ Apache Storm

เกี่ยวกับ Storm

Storm ถูกสร้างขึ้นโดย Nathan Marz และทีมงานที่ BackType ในช่วงเวลาสั้น ๆ Apache Storm กลายเป็นมาตรฐานสำหรับระบบประมวลผลแบบเรียลไทม์แบบกระจายซึ่งช่วยให้คุณประมวลผลข้อมูลจำนวนมหาศาลได้ สตอร์มเร็วมากและค่ามาตรฐานโอเวอร์คล็อกด้วยการประมวลผลมากกว่าหนึ่งล้านทูเปิลต่อวินาทีต่อโหนด Apache Storm ทำงานอย่างต่อเนื่องโดยใช้ข้อมูลจากแหล่งที่กำหนด (Spouts) และส่งข้อมูลไปยังท่อประมวลผล (Bolts) Com-bined Spouts และ Bolts สร้างโทโพโลยี

บูรณาการกับ Storm

Kafka และ Storm เสริมซึ่งกันและกันอย่างเป็นธรรมชาติและความร่วมมืออันทรงพลังของพวกเขาช่วยให้การวิเคราะห์สตรีมมิ่งแบบเรียลไทม์สำหรับข้อมูลขนาดใหญ่ที่เคลื่อนไหวอย่างรวดเร็ว การรวม Kafka และ Storm ช่วยให้นักพัฒนาสามารถนำเข้าและเผยแพร่สตรีมข้อมูลจากโครงสร้าง Storm

กระแสความคิด

พวยกาเป็นแหล่งที่มาของลำธาร ตัวอย่างเช่นพวยกาอาจอ่านสิ่งที่เกิดขึ้นจากหัวข้อคาฟคาและส่งออกเป็นสตรีม โบลต์ใช้สตรีมอินพุตกระบวนการและอาจส่งกระแสข้อมูลใหม่ Bolts สามารถทำอะไรก็ได้ตั้งแต่การเรียกใช้ฟังก์ชั่นการกรองสิ่งที่เพิ่มขึ้นทำการสตรีมการรวมการรวมการสตรีมการพูดคุยกับฐานข้อมูลและอื่น ๆ แต่ละโหนดในโทโพโลยี Storm ทำงานแบบขนาน โทโพโลยีทำงานไปเรื่อย ๆ จนกว่าคุณจะยุติ Storm จะมอบหมายงานที่ล้มเหลวโดยอัตโนมัติ นอกจากนี้สตอร์มยังรับประกันว่าจะไม่มีการสูญหายของข้อมูลแม้ว่าเครื่องจะล่มและข้อความหลุดก็ตาม

ให้เราดูรายละเอียดเกี่ยวกับ API การรวม Kafka-Storm มีสามคลาสหลักในการรวม Kafka กับ Storm มีดังนี้ -

BrokerHosts - ZkHosts & StaticHosts

BrokerHosts เป็นอินเทอร์เฟซและ ZkHosts และ StaticHosts เป็นสองการใช้งานหลัก ZkHosts ใช้เพื่อติดตามโบรกเกอร์ Kafka แบบไดนามิกโดยการรักษารายละเอียดใน ZooKeeper ในขณะที่ StaticHosts ใช้เพื่อตั้งค่าโบรกเกอร์ Kafka และรายละเอียดด้วยตนเอง / แบบคงที่ ZkHosts เป็นวิธีที่ง่ายและรวดเร็วในการเข้าถึงโบรกเกอร์ Kafka

ลายเซ็นของ ZkHosts มีดังต่อไปนี้ -

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

โดยที่นายหน้า ZkStr คือโฮสต์ ZooKeeper และนายหน้า ZkPath คือเส้นทาง ZooKeeper เพื่อรักษารายละเอียดนายหน้าของคาฟคา

KafkaConfig API

API นี้ใช้เพื่อกำหนดการตั้งค่าคอนฟิกสำหรับคลัสเตอร์ Kafka ลายเซ็นของ Kafka Con-fig ถูกกำหนดไว้ดังนี้

public KafkaConfig(BrokerHosts hosts, string topic)

Hosts - BrokerHosts สามารถเป็น ZkHosts / StaticHosts

Topic - ชื่อหัวข้อ

SpoutConfig API

Spoutconfig เป็นส่วนขยายของ KafkaConfig ที่รองรับข้อมูล ZooKeeper เพิ่มเติม

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

Hosts - BrokerHosts สามารถใช้งานอินเทอร์เฟซ BrokerHosts ใดก็ได้
Topic - ชื่อหัวข้อ
zkRoot - เส้นทางราก ZooKeeper
id −พวยกาเก็บสถานะของการชดเชยที่บริโภคใน Zookeeper รหัสควรระบุพวยกาของคุณโดยไม่ซ้ำกัน

SchemeAsMultiScheme

SchemeAsMultiScheme เป็นอินเทอร์เฟซที่กำหนดวิธีการที่ ByteBuffer ที่ใช้จาก Kafka จะถูกเปลี่ยนเป็น Storm tuple ได้มาจาก MultiScheme และยอมรับการใช้งานคลาส Scheme มีการใช้งานคลาส Scheme จำนวนมากและการใช้งานแบบนั้นคือ StringScheme ซึ่งแยกวิเคราะห์ไบต์เป็นสตริงธรรมดา นอกจากนี้ยังควบคุมการตั้งชื่อฟิลด์เอาต์พุตของคุณ ลายเซ็นถูกกำหนดไว้ดังนี้

public SchemeAsMultiScheme(Scheme scheme)

Scheme - บัฟเฟอร์ไบต์ที่บริโภคจากคาฟคา

KafkaSpout API

KafkaSpout คือการใช้งานพวยกาของเราซึ่งจะทำงานร่วมกับ Storm มันดึง mes-sages จากหัวข้อ kafka และส่งมันไปยังระบบนิเวศของ Storm ในรูปแบบ tuples KafkaSpout รับรายละเอียดการกำหนดค่าจาก SpoutConfig

ด้านล่างนี้เป็นโค้ดตัวอย่างสำหรับสร้าง Kafka spout แบบง่ายๆ

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

การสร้างกลอน

Bolt เป็นส่วนประกอบที่ใช้ tuples เป็นอินพุตประมวลผลทูเพิลและสร้างสิ่งทอปเปิลใหม่เป็นเอาต์พุต Bolts จะใช้อินเทอร์เฟซ IRichBolt ในโปรแกรมนี้มีการใช้โบลต์สองคลาส WordSplitter-Bolt และ WordCounterBolt เพื่อดำเนินการ

อินเทอร์เฟซ IRichBolt มีวิธีการดังต่อไปนี้ -

Prepare- จัดเตรียมโบลต์พร้อมสภาพแวดล้อมในการดำเนินการ ตัวดำเนินการจะเรียกใช้วิธีนี้เพื่อเริ่มต้นพวยกา
Execute - ประมวลผลอินพุตทูเพิลเดียว
Cleanup - เรียกว่าเมื่อสายฟ้ากำลังจะปิดลง
declareOutputFields - ประกาศสคีมาผลลัพธ์ของทูเปิล

ให้เราสร้าง SplitBolt.java ซึ่งใช้ตรรกะในการแยกประโยคออกเป็นคำและ CountBolt.java ซึ่งใช้ตรรกะในการแยกคำที่ไม่ซ้ำกันและนับจำนวนที่เกิดขึ้น

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

ส่งไปยังโทโพโลยี

โทโพโลยีแบบพายุนั้นเป็นโครงสร้างแบบ Thrift คลาส TopologyBuilder มีวิธีการที่ง่ายและสะดวกในการสร้างโทโพโลยีที่ซับซ้อน คลาส TopologyBuilder มีเมธอดในการตั้งพวยกา (setSpout) และตั้งค่าโบลต์ (setBolt) ในที่สุด TopologyBuilder ได้ createTopology เพื่อสร้าง to-pology วิธี shuffleGrouping และ fieldsGrouping ช่วยในการตั้งค่าการจัดกลุ่มสตรีมสำหรับพวยกาและสลักเกลียว

Local Cluster- เพื่อวัตถุประสงค์ในการพัฒนาเราสามารถสร้างคลัสเตอร์ท้องถิ่นโดยใช้LocalClusterวัตถุแล้วส่ง topology โดยใช้submitTopologyวิธีการLocalClusterระดับ

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

ก่อนที่จะย้ายการรวบรวมการรวม Kakfa-Storm ต้องการไลบรารี java ไคลเอนต์ ZooKeeper ผู้ดูแล ผู้ดูแลเวอร์ชัน 2.9.1 รองรับ Apache Storm เวอร์ชัน 0.9.5 (ซึ่งเราใช้ในบทช่วยสอนนี้) โหลดไฟล์ jar ที่ระบุด้านล่างลงและวางไว้ในพา ธ คลาส java

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

หลังจากรวมไฟล์อ้างอิงแล้วให้คอมไพล์โปรแกรมโดยใช้คำสั่งต่อไปนี้

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

การดำเนินการ

เริ่ม Kafka Producer CLI (อธิบายในบทก่อนหน้า) สร้างหัวข้อใหม่ชื่อmy-first-topicและให้ข้อความตัวอย่างตามที่แสดงด้านล่าง -

hello
kafka
storm
spark
test message
another test message

ตอนนี้เรียกใช้แอปพลิเคชันโดยใช้คำสั่งต่อไปนี้ -

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

ตัวอย่างผลลัพธ์ของแอปพลิเคชันนี้ระบุไว้ด้านล่าง -

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