Apache Kafka -7 與Storm整合
Apache Kafka教程 之 與Storm整合
Apache Kafka - 與Storm整合
關於Storm
Storm最初是由Nathan Marz和BackType建立的。在短時間內，Apache Storm成為分散式實時處理系統的標準，可讓您處理大量資料。Storm非常快，每個節點每秒處理超過一百萬個元組的基準測試。Apache Storm持續執行，從配置的源（Spouts）中消耗資料，並將資料傳遞到處理管道（Bolts）。組合，spout和Bolts 做拓撲。
與Storm融合
Kafka和Storm自然互補，他們強大的合作使得實時流式分析能夠快速移動大資料。Kafka和Storm整合是讓開發人員更容易地從Storm拓撲中吸收和釋出資料流。
概念流
spout是流的來源。例如，一個spout可以從Kafka主題中讀取元組，並將其作為流發出。Bolts 消耗輸入流，處理並可能發射新的流。Bolts 可以從執行功能，過濾元組，進行流聚合，流式連線，與資料庫交談等操作。Storm拓撲中的每個節點並行執行。拓撲執行無限期，直到您終止它。Storm將自動重新分配任何失敗的任務。此外，Storm保證不會丟失資料，即使機器掉線，資訊丟失。
讓我們詳細介紹一下Kafka-Storm整合API。將卡夫卡與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 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 -spout stores儲存在Zookeeper中消耗的偏移量的狀態。該ID應該唯一標識您的spout。
SchemeAsMultiScheme
SchemeAsMultiScheme是一個介面，用於指示從Kafka消耗的ByteBuffer如何轉換為Storm元組。它來自MultiScheme並接受Scheme類的實現。有很多的Scheme類的實現，一個這樣的實現是StringScheme，它將位元組解析為一個簡單的字串。它還控制輸出欄位的命名。簽名的定義如下。
public SchemeAsMultiScheme(Scheme scheme)
- Scheme - 從卡夫卡消費的位元組緩衝區。
KafkaSpout API
KafkaSpout是我們的spout實現，它將與Storm整合。它從卡夫卡主題中獲取訊息，並將其作為元組傳送到Storm生態系統。KafkaSpout從SpoutConfig獲取其配置詳細資訊。
以下是建立簡單Kafkaspout的示例程式碼。
// 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 Creation
Bolt是一個元件，它將元組作為輸入，處理元組，併產生新的元組作為輸出。Bolts 將實現IRichBolt介面。在這個程式中，使用兩個Bolts 類WordSplitter-Bolt和WordCounterBolt進行操作。
IRichBolt介面有以下幾種方法 -
- Prepare- 為Bolts 提供執行的環境。執行器將執行此方法來初始化spout。
- Execute- 處理一個單獨的輸入元組。
- Cleanup - 當Bolts 關閉時呼叫。
- 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;
}
}提交到Topology
Storm拓撲基本上是一個節儉的結構。TopologyBuilder類提供簡單而簡單的方法來建立複雜的拓撲。TopologyBuilder類具有設定spout（setSpout）和設定bolt（setBolt）的方法。最後，TopologyBuilder已經建立了一個建立To-pology的拓撲。shuffleGrouping和fieldsGrouping方法有助於為噴嘴和Bolts 設定流分組。
本地叢集 - 為了開發目的，我們可以使用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庫。策展人版本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