Kafka與Storm整合

在本章中，我們將學習如何將Kafka與Apache Storm集成。

Storm是什麼？

Storm最初是由Nathan Marz和BackType團隊創建的。在很短的時間內，Apache Storm成爲分佈式實時處理系統的標準，用於處理大數據。 Storm速度非常快，每個節點每秒處理超過一百萬個元組的基準時鐘。 Apache Storm持續運行，從配置的源(Spouts)中消耗數據並將數據傳遞到處理管道(Bolts)。組合 Spouts 和 Bolts 構成一個拓撲。

與Storm整合

Kafka和Storm自然而然地相互補充，它們強大的合作能夠實現快速移動大數據的實時流式分析。 Kafka和Storm的整合使得開發者更容易從Storm拓撲中獲取和發佈數據流。

概念流程

噴口(spout)是流的來源。例如，spout可能會讀取卡夫卡主題中的元組並將其作爲流發送。 Bolts消耗輸入流，處理並可能發射新的流。 Bolts可以做任何事情，從運行功能，過濾元組，流聚合，流式連接，與數據庫交互等等。 Storm拓撲中的每個節點並行執行。一個拓撲無限期地運行，直到終止它。 Storm會自動重新分配任何失敗的任務。此外，即使機器停機並且信息丟失，Storm也可以保證不會丟失數據。

下面來看看Kafka-Storm集成API。有三個主要類將Kafka和Storm結合在一起。他們如下 -

BrokerHosts - ZkHosts＆StaticHosts

BrokerHosts是一個接口，ZkHosts和StaticHosts是它的兩個主要實現。 ZkHosts用於通過在ZooKeeper中維護詳細信息來動態跟蹤Kafka經紀人，而StaticHosts用於手動/靜態設置Kafka經紀人及其詳細信息。 ZkHosts是訪問Kafka經紀人的簡單而快捷的方式。

ZkHosts的簽名如下 -

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

其中brokerZkStr是ZooKeeper主機，brokerZkPath是維護Kafka代理細節的ZooKeeper路徑。

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 - spout存儲在Zookeeper中消耗的偏移量的狀態。該ID應該唯一標識的spout。

SchemeAsMultiScheme

SchemeAsMultiScheme是一個接口，它規定了從Kafka消耗的ByteBuffer如何轉換爲 storm 元組。它來自MultiScheme並接受Scheme類的實現。Scheme類有很多實現，一個這樣的實現是StringScheme，它將字節解析爲一個簡單的字符串。它還控制輸出字段的命名。簽名定義如下。

public SchemeAsMultiScheme(Scheme scheme)

參數

scheme - 從kafka消耗的字節緩衝區。

KafkaSpout API

KafkaSpout是spout實現，它將與Storm整合。它從kafka主題獲取消息並將其作爲元組發送到Storm生態系統中。 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創建

Bolt是一個將元組作爲輸入，處理元組並生成新的元組作爲輸出的組件。 Bolts將實現IRichBolt接口。在這個程序中，使用兩個類 - WordSplitter-Bolt和WordCounterBolt來執行操作。

IRichBolt接口有以下方法 -

prepare - 爲 bolt 提供執行的環境。執行者將運行此方法來初始化spout。
prepare - 處理輸入的單個元組。
prepare - 當bolt即將關閉時調用。
declareOutputFields - 聲明元組的輸出模式。

下面創建一個Java文件: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;
   }
}

提交到拓撲

Storm拓撲基本上是一個Thrift結構。 TopologyBuilder類提供了簡單而簡單的方法來創建複雜的拓撲。 TopologyBuilder類具有設置spout (setSpout)和設置bolt(setBolt)的方法。最後，TopologyBuilder使用createTopology()來創建拓樸學。 shuffleGrouping和fieldsGrouping方法有助於設置spout和bolt的流分組。

本地羣集 - 出於開發目的，我們可以使用LocalCluster對象創建本地羣集，然後使用LocalCluster類的submitTopology方法提交拓撲。

文件: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集成需要館長ZooKeeper客戶端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

Kafka教程

Spring Kafka實例

Storm是什麼？

與Storm整合

Bolt創建

提交到拓撲