Apache Kafka - 與Storm整合

上一頁
下一頁

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

關於Storm

Storm最初由BackType的Nathan Marz和團隊建立。在很短的時間內，Apache Storm成為了分散式即時處理系統的標準，它允許你處理海量資料。Storm非常快，基準測試表明它每個節點每秒可以處理超過一百萬個元組。Apache Storm持續執行，從配置的資料來源（Spout）中獲取資料，並將資料傳遞到處理管道（Bolt）中。Spout和Bolt組合在一起構成一個拓撲結構。

與Storm整合

Kafka和Storm自然地相互補充，它們強大的合作使快速移動的大資料的即時流分析成為可能。Kafka和Storm的整合是為了使開發人員更容易地從Storm拓撲結構中攝取和釋出資料流。

概念流程

Spout是資料流的來源。例如，Spout可以從Kafka主題讀取元組並將其作為資料流發出。Bolt消費輸入流，處理並可能發出新的資料流。Bolt可以執行任何操作，從執行函式、過濾元組、執行流聚合、流連線、與資料庫互動等等。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是ZooKeeper路徑，用於維護Kafka代理詳細資訊。

KafkaConfig API

此API用於定義Kafka叢集的配置設定。Kafka Config的簽名定義如下

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是一個元件，它以元組作為輸入，處理元組，併產生新的元組作為輸出。Bolt將實現IRichBolt介面。在這個程式中，使用兩個bolt類WordSplitterBolt和WordCounterBolt來執行操作。

IRichBolt介面具有以下方法：

  • Prepare - 為bolt提供執行環境。執行器將執行此方法來初始化spout。

  • Execute - 處理單個輸入元組。

  • Cleanup - 當bolt即將關閉時呼叫。

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

提交到拓撲結構

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整合需要curator ZooKeeper客戶端Java庫。Curator版本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
列印頁面
上一頁
下一頁
廣告

© . All rights reserved.