Apache Kafka 简明教程
Apache Kafka - Integration With Storm
在本章中,我们将学习如何集 Kafka 与 Apache Storm。
About Storm
Storm 最初由 Nathan Marz 和 BackType 团队创建。在很短的时间内,Apache Storm 已成为分布式实时处理系统的标准,允许你处理海量数据。Storm 非常快,基准测试证明每个节点每秒可处理超过一百万个元组。Apache Storm 连续运行,从配置的源(注水口)消耗数据,并将数据传递到处理管线(螺栓)。结合注水口和螺栓形成拓扑。
Integration with Storm
Kafka 和 Storm 会自然地相互补充,而且它们功能强大的合作可为快速移动的大数据启用实时流分析。Kafka 和 Storm 集成更便于开发人员从 Storm 拓扑中提取和发布数据流。
Conceptual flow
注水口是流的来源。例如,注水口可能会从 Kafka 主题读取元组并以流形式发出它们。螺栓会消耗输入流、处理并可能发出新的流。螺栓的操作从运行函数、过滤元组到执行流聚合、流联接、与数据库通信等可谓无所不包。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 配置的签名定义如下
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 − topic name.
-
zkRoot − ZooKeeper 根路径。
-
id − 分流将它消耗的偏移量状态存储在 Zookeeper 中。id 应唯一标识你的分流。
SchemeAsMultiScheme
SchemeAsMultiScheme 是一个接口,它规定从 Kafka 消费的 ByteBuffer 转换为 storm 元组的方式。它源自 MultiScheme 并接受 Scheme 类的实现。Scheme 类的实现有很多,其中一个实现是 StringScheme,它将字节解析为简单的字符串。它还控制着输出字段的命名。签名定义如下所示。
public SchemeAsMultiScheme(Scheme scheme)
-
Scheme − 从 kafka 消费的字节缓冲区。
KafkaSpout API
KafkaSpout 是我们的流嘴实现,它将与 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 Creation
Bolt 是一个将元组作为输入,处理元组并生成新元组作为输出的组件。Bolt 将实现 IRichBolt 接口。在此程序中,Bolt 类 WordSplitter-Bolt 和 WordCounterBolt 用于执行操作。
IRichBolt 接口具有以下方法:
-
Prepare − 为 Bolt 提供执行环境。执行器将运行此方法来初始化流嘴。
-
Execute − 处理单个输入元组。
-
Cleanup − 在 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()+" : " + entry.getValue());
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
@Override
public Map<String, Object> getComponentConfiguration() {
return null;
}
}
Submitting to Topology
Storm 拓扑基本上是一个 Thrift 结构。TopologyBuilder 类提供了简单易用的方法来创建复杂的拓扑。TopologyBuilder 类具有设置分流 (setSpout) 和设置 Bolt (setBolt) 的方法。最后,TopologyBuilder 具有 createTopology 来创建拓扑。shuffleGrouping 和 fieldsGrouping 方法有助于设置流嘴和 Bolt 的流分组。
Local Cluster − 出于开发目的,我们可以使用 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 集成需要 curator ZooKeeper client 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
Execution
启动 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