Apache Kafka 与 Storm 的集成
在本章中，我们将学习如何将 Kafka 与 Apache Storm 集成。
关于Storm
Storm 最初由 Nathan Marz 和 BackType 的团队创建。在很短的时间内，Apache Storm 成为分布式实时处理系统的标准，允许你处理大量数据。 Storm 速度非常快，基准测试显示每个节点每秒处理超过一百万个元组。 Apache Storm 持续运行，从配置的源 (Spouts) 消耗数据并将数据向下传递到处理管道 (Bolts)。 Spout 和 Bolts 组合在一起构成了一个拓扑。
与Storm集成
Kafka 和 Storm 自然相得益彰，强大的合作为快速移动的大数据提供实时流式分析。 Kafka 和 Storm 的集成是为了让开发人员更容易从 Storm 拓扑中摄取和发布数据流。
概念流程
喷口是流的来源。例如，spout 可能会从 Kafka 主题中读取元组并将它们作为流发出。螺栓消耗输入流，处理并可能发出新流。 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 是 ZooKeeper 路径，用于维护 Kafka 代理详细信息。
Kafka配置 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 是一个接口，它指示从 Kafka 消耗的 ByteBuffer 如何转换为风暴元组。它派生自 MultiScheme 并接受 Scheme 类的实现。 Scheme 类有很多实现，其中一个实现是 StringScheme，它将字节解析为简单的字符串。它还控制输出字段的命名。签名定义如下。
public SchemeAsMultiScheme(Scheme scheme)
Scheme： 从 kafka 消耗的字节缓冲区。
KafkaSpout API
KafkaSpout 是我们的 spout 实现，它将与 Storm 集成。它从 kafka 主题中获取消息，并将其作为元组发送到 Storm 生态系统中。 KafkaSpout 从 SpoutConfig 获取其配置详细信息。
下面是创建简单 Kafka spout 的示例代码。
// ZooKeeper 连接字符串 BrokerHosts hosts = new ZkHosts(zkConnString); // 创建 SpoutConfig 对象 SpoutConfig spoutConfig = new SpoutConfig(hosts, topicName, "/" + topicName UUID.randomUUID().toString()); // 将 ByteBuffer 转换为字符串。 spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme()); // 将 SpoutConfig 分配给 KafkaSpout。 KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
创建Bolts
Bolt 是一个将元组作为输入，处理元组，并产生新元组作为输出的组件。 Bolts 将实现 IRichBolt 接口。在这个程序中，两个螺栓类 WordSplitter-Bolt 和 WordCounterBolt 用于执行操作。
IRichBolt 接口有以下方法：
Prepare： 为bolt提供一个执行环境。执行程序将运行此方法来初始化 spout。
Execute：处理输入的单个元组。
Cleanup： 当一个bolt要关闭时调用。
声明输出字段：声明元组的输出模式。
让我们创建 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 和 bolts 设置流分组。
本地集群：出于开发目的，我们可以使用创建本地集群对象，然后使用提交拓扑的方法class.
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(在上一章解释过)，创建一个名为并提供一些示例消息，如下所示：
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