Apache Kafka 整合 Storm

storm最初由nathan marz和backtype的团队创建。在短时间内，apache storm成为分布式实时处理系统的标准，允许您处理大量数据。 storm是非常快的，并且一个基准时钟为每个节点每秒处理超过一百万个元组。 apache storm持续运行，从配置的源(spouts)消耗数据，并将数据传递到处理管道(bolts)。联合，spouts和bolt构成一个拓扑。

与storm集成

kafka和storm自然互补，它们强大的合作能够实现快速移动的大数据的实时流分析。 kafka和storm集成是为了使开发人员更容易地从storm拓扑获取和发布数据流。

概念流

spouts是流的源。例如，一个喷头可以从kafka topic读取元组并将它们作为流发送。 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 con-fig的签名定义如下

public kafkaconfig(brokerhosts hosts, string topic)

主机 - brokerhosts可以是zkhosts / statichosts。

主题 - 主题名称。

spoutconfig api

spoutconfig是kafkaconfig的扩展，支持额外的zookeeper信息。

public spoutconfig(brokerhosts hosts, string topic, string zkroot, string id)

主机 - brokerhosts可以是brokerhosts接口的任何实现
主题 - 主题名称。
zkroot - zookeeper根路径。
id - spouts存储在zookeeper中消耗的偏移量的状态。 id应该唯一标识您的喷嘴。

schemeasmultischeme

schemeasmultischeme是一个接口，用于指示如何将从kafka中消耗的bytebuffer转换为风暴元组。它源自multischeme并接受scheme类的实现。有很多scheme类的实现，一个这样的实现是stringscheme，它将字节解析为一个简单的字符串。它还控制输出字段的命名。签名定义如下。

public schemeasmultischeme(scheme scheme)

方案 - 从kafka消耗的字节缓冲区。

kafkaspout api

kafkaspout是我们的spout实现，它将与storm集成。它从kafka主题获取消息，并将其作为元组发送到storm生态系统。 kafkaspout从spoutconfig获取其配置详细信息。

下面是一个创建一个简单的kafka喷水嘴的示例代码。

// 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类wordsplitter-bolt和wordcounterbolt来执行操作。

irichbolt接口有以下方法 -

准备 - 为bolt提供要执行的环境。执行器将运行此方法来初始化喷头。
执行 - 处理单个元组的输入。
清理 - 当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()&plus;" : " &plus; 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来创建to-pology。 shufflegrouping和fieldsgrouping方法有助于为喷头和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库。策展人版本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