📅  最后修改于: 2020-12-03 02:59:24             🧑  作者: Mango
在本章中,我们将学习如何将Kafka与Apache Storm集成。
Storm最初由BackType的Nathan Marz及其团队创建。在短时间内,Apache Storm成为分布式实时处理系统的标准,该系统使您可以处理大量数据。 Storm的速度非常快,基准测试表明它每秒可处理每个节点超过一百万个元组。 Apache Storm连续运行,使用来自已配置源(Spouts)的数据,并将数据向下传递到处理管道(Bolts)。喷头和螺栓结合在一起构成一个拓扑。
Kafka和Storm自然互补,它们的强大合作可实现实时流分析,以快速移动大数据。 Kafka和Storm的集成将使开发人员更轻松地从Storm拓扑中提取和发布数据流。
壶嘴是溪流的源头。例如,喷口可能会从Kafka主题中读取元组并将其作为流发出。螺栓消耗输入流,处理并可能发出新流。螺栓可以执行任何操作,包括运行函数,过滤元组,执行流聚合,流连接,与数据库对话等等。 Storm拓扑中的每个节点都并行执行。拓扑无限期运行,直到您终止它。 Storm将自动重新分配所有失败的任务。此外,Storm保证即使机器宕机和消息丢失也不会丢失数据。
让我们详细了解Kafka-Storm集成API。有三个主要的类将Kafka与Storm集成。它们如下-
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经纪人详细信息。
该API用于定义Kafka集群的配置设置。 Kafka Con-fig的签名定义如下
public KafkaConfig(BrokerHosts hosts, string topic)
主机-BrokerHost可以是ZkHosts / StaticHosts。
主题-主题名称。
Spoutconfig是KafkaConfig的扩展,支持其他ZooKeeper信息。
public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)
主机-BrokerHosts可以是BrokerHosts接口的任何实现
主题-主题名称。
zkRoot -ZooKeeper根路径。
id-喷嘴将其消耗的偏移状态存储在Zookeeper中。该ID应该唯一标识您的壶嘴。
SchemeAsMultiScheme是一个接口,该接口指示从Kafka消耗的ByteBuffer如何转换为风暴元组。它从MultiScheme派生并接受Scheme类的实现。 Scheme类有很多实现,其中一种实现是StringScheme,它将字节解析为简单的字符串。它还控制输出字段的命名。签名定义如下。
public SchemeAsMultiScheme(Scheme scheme)
方案-从卡夫卡消耗的字节缓冲区。
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是一个将元组作为输入,处理该元组并生成新的元组作为输出的组件。螺栓将实现IRichBolt接口。在此程序中,两个螺栓类WordSplitter-Bolt和WordCounterBolt用于执行操作。
IRichBolt接口具有以下方法-
准备-为螺栓提供执行环境。执行者将运行此方法来初始化喷口。
执行-处理单个输入元组。
清理-当螺栓即将关闭时调用。
clarifyOutputFields-声明元组的输出模式。
让我们创建SplitBolt.java和CountBolt.java,它实现将句子拆分成单词的逻辑,而CountBolt.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 getComponentConfiguration() {
return null;
}
}
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 counters;
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context,
OutputCollector collector) {
this.counters = new HashMap();
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 entry:counters.entrySet()){
System.out.println(entry.getKey()+" : " + entry.getValue());
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
@Override
public Map getComponentConfiguration() {
return null;
}
}
Storm拓扑基本上是Thrift结构。 TopologyBuilder类提供了创建复杂拓扑的简单方法。 TopologyBuilder类具有设置喷口(setSpout)和设置螺栓(setBolt)的方法。最后,TopologyBuilder具有createTopology来创建拓扑。 shuffleGrouping和fieldsGrouping方法有助于为喷口和螺栓设置流分组。
本地群集-出于开发目的,我们可以使用LocalCluster
对象创建本地群集,然后使用LocalCluster
类的SubmitTopology
方法提交拓扑。
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库。 Curator版本2.9.1支持Apache Storm版本0.9.5(我们在本教程中使用)。下载以下指定的jar文件并将其放置在java类路径中。
包含依赖项文件后,使用以下命令编译程序,
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