在本章中，我们将学习如何将kafka与apache storm集成。
storm最初是由nathan marz和backtype团队创建的。 在很短的时间内，apache storm成为分布式实时处理系统的标准，用于处理大数据。 storm速度非常快，每个节点每秒处理超过一百万个元组的基准时钟。 apache storm持续运行，从配置的源(spouts)中消耗数据并将数据传递到处理管道(bolts)。 组合 spouts 和 bolts 构成一个拓扑。
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)
参数
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是一个将元组作为输入，处理元组并生成新的元组作为输出的组件。 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.jarcurator-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
