Apache Kafka

Apache Kafka教程之与Storm集成

http://blogxinxiucan.sh1.newtouch.com/

原文地址： http://blogxinxiucan.sh1.newtouch.com/2017/07/13/Apache-Kafka-与Storm集成/

Apache Kafka - 与Storm集成

关于Storm

Storm最初是由Nathan Marz和BackType创建的。在短时间内，Apache Storm成为分布式实时处理系统的标准，可让您处理大量数据。Storm非常快，每个节点每秒处理超过一百万个元组的基准测试。Apache Storm持续运行，从配置的源（Spouts）中消耗数据，并将数据传递到处理管道（Bolts）。组合，spout和Bolts 做拓扑。

与Storm融合

Kafka和Storm自然互补，他们强大的合作使得实时流式分析能够快速移动大数据。Kafka和Storm集成是让开发人员更容易地从Storm拓扑中吸收和发布数据流。

概念流
spout是流的来源。例如，一个spout可以从Kafka主题中读取元组，并将其作为流发出。Bolts 消耗输入流，处理并可能发射新的流。Bolts 可以从运行功能，过滤元组，进行流聚合，流式连接，与数据库交谈等操作。Storm拓扑中的每个节点并行执行。拓扑运行无限期，直到您终止它。Storm将自动重新分配任何失败的任务。此外，Storm保证不会丢失数据，即使机器掉线，信息丢失。

让我们详细介绍一下Kafka-Storm集成API。将卡夫卡与Storm整合有三个主要课程。他们如下 -

BrokerHosts - ZkHosts＆StaticHosts
BrokerHosts是一个接口，ZkHosts和StaticHosts是它的两个主要实现。ZkHosts用于通过维护ZooKeeper中的细节来动态跟踪Kafka经纪人，而StaticHosts用于手动/静态设置Kafka经纪人及其详细信息。ZkHosts是访问Kafka经纪人的简单而快速的方式。

ZkHosts的签名如下：

1public ZkHosts(String brokerZkStr, String brokerZkPath)
2public ZkHosts(String brokerZkStr)

其中brokerZkStr是ZooKeeper主机和brokerZkPath是ZooKeeper路径来维护Kafka代理详细信息。

KafkaConfig API
此API用于定义Kafka集群的配置设置。Kafka Con-fig的签名定义如下

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 stores存储在Zookeeper中消耗的偏移量的状态。该ID应该唯一标识您的spout。

SchemeAsMultiScheme
SchemeAsMultiScheme是一个接口，用于指示从Kafka消耗的ByteBuffer如何转换为Storm元组。它来自MultiScheme并接受Scheme类的实现。有很多的Scheme类的实现，一个这样的实现是StringScheme，它将字节解析为一个简单的字符串。它还控制输出字段的命名。签名的定义如下。

public SchemeAsMultiScheme(Scheme scheme)

Scheme - 从卡夫卡消费的字节缓冲区。

KafkaSpout API
KafkaSpout是我们的spout实现，它将与Storm集成。它从卡夫卡主题中获取消息，并将其作为元组发送到Storm生态系统。KafkaSpout从SpoutConfig获取其配置详细信息。

以下是创建简单Kafkaspout的示例代码。

1// ZooKeeper connection string
2BrokerHosts hosts = new ZkHosts(zkConnString);
3
4//Creating SpoutConfig Object
5SpoutConfig spoutConfig = new SpoutConfig(hosts, 
6   topicName, "/" + topicName UUID.randomUUID().toString());
7
8//convert the ByteBuffer to String.
9spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
10
11//Assign SpoutConfig to KafkaSpout.
12KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

Bolt Creation

Bolt是一个组件，它将元组作为输入，处理元组，并产生新的元组作为输出。Bolts 将实现IRichBolt接口。在这个程序中，使用两个Bolts 类WordSplitter-Bolt和WordCounterBolt进行操作。

IRichBolt接口有以下几种方法 -

Prepare- 为Bolts 提供执行的环境。执行器将运行此方法来初始化spout。
Execute- 处理一个单独的输入元组。
Cleanup - 当Bolts 关闭时调用。
declareOutputFields - 声明元组的输出模式。

让我们创建SplitBolt.java，它实现将一个句子分割成单词和CountBolt.java的逻辑，该实现逻辑用于分隔唯一的单词并计算其发生。

SplitBolt.java

1import java.util.Map;
2
3import backtype.storm.tuple.Tuple;
4import backtype.storm.tuple.Fields;
5import backtype.storm.tuple.Values;
6
7import backtype.storm.task.OutputCollector;
8import backtype.storm.topology.OutputFieldsDeclarer;
9import backtype.storm.topology.IRichBolt;
10import backtype.storm.task.TopologyContext;
11
12public class SplitBolt implements IRichBolt {
13   private OutputCollector collector;
14   
15   @Override
16   public void prepare(Map stormConf, TopologyContext context,
17      OutputCollector collector) {
18      this.collector = collector;
19   }
20   
21   @Override
22   public void execute(Tuple input) {
23      String sentence = input.getString(0);
24      String[] words = sentence.split(" ");
25      
26      for(String word: words) {
27         word = word.trim();
28         
29         if(!word.isEmpty()) {
30            word = word.toLowerCase();
31            collector.emit(new Values(word));
32         }
33         
34      }
35
36      collector.ack(input);
37   }
38   
39   @Override
40   public void declareOutputFields(OutputFieldsDeclarer declarer) {
41      declarer.declare(new Fields("word"));
42   }
43
44   @Override
45   public void cleanup() {}
46   
47   @Override
48   public Map<String, Object> getComponentConfiguration() {
49      return null;
50   }
51   
52}

CountBolt.java

1import java.util.Map;
2import java.util.HashMap;
3
4import backtype.storm.tuple.Tuple;
5import backtype.storm.task.OutputCollector;
6import backtype.storm.topology.OutputFieldsDeclarer;
7import backtype.storm.topology.IRichBolt;
8import backtype.storm.task.TopologyContext;
9
10public class CountBolt implements IRichBolt{
11   Map<String, Integer> counters;
12   private OutputCollector collector;
13   
14   @Override
15   public void prepare(Map stormConf, TopologyContext context,
16   OutputCollector collector) {
17      this.counters = new HashMap<String, Integer>();
18      this.collector = collector;
19   }
20
21   @Override
22   public void execute(Tuple input) {
23      String str = input.getString(0);
24      
25      if(!counters.containsKey(str)){
26         counters.put(str, 1);
27      }else {
28         Integer c = counters.get(str) +1;
29         counters.put(str, c);
30      }
31   
32      collector.ack(input);
33   }
34
35   @Override
36   public void cleanup() {
37      for(Map.Entry<String, Integer> entry:counters.entrySet()){
38         System.out.println(entry.getKey()+" : " + entry.getValue());
39      }
40   }
41
42   @Override
43   public void declareOutputFields(OutputFieldsDeclarer declarer) {
44   
45   }
46
47   @Override
48   public Map<String, Object> getComponentConfiguration() {
49      return null;
50   }
51}

提交到Topology

Storm拓扑基本上是一个节俭的结构。TopologyBuilder类提供简单而简单的方法来创建复杂的拓扑。TopologyBuilder类具有设置spout（setSpout）和设置bolt（setBolt）的方法。最后，TopologyBuilder已经创建了一个创建To-pology的拓扑。shuffleGrouping和fieldsGrouping方法有助于为喷嘴和Bolts 设置流分组。

本地集群 - 为了开发目的，我们可以使用LocalCluster对象创建本地集群，然后使用LocalCluster类的submitTopology方法提交拓扑。

KafkaStormSample.java

1import backtype.storm.Config;
2import backtype.storm.LocalCluster;
3import backtype.storm.topology.TopologyBuilder;
4
5import java.util.ArrayList;
6import java.util.List;
7import java.util.UUID;
8
9import backtype.storm.spout.SchemeAsMultiScheme;
10import storm.kafka.trident.GlobalPartitionInformation;
11import storm.kafka.ZkHosts;
12import storm.kafka.Broker;
13import storm.kafka.StaticHosts;
14import storm.kafka.BrokerHosts;
15import storm.kafka.SpoutConfig;
16import storm.kafka.KafkaConfig;
17import storm.kafka.KafkaSpout;
18import storm.kafka.StringScheme;
19
20public class KafkaStormSample {
21   public static void main(String[] args) throws Exception{
22      Config config = new Config();
23      config.setDebug(true);
24      config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
25      String zkConnString = "localhost:2181";
26      String topic = "my-first-topic";
27      BrokerHosts hosts = new ZkHosts(zkConnString);
28      
29      SpoutConfig kafkaSpoutConfig = new SpoutConfig (hosts, topic, "/" + topic,    
30         UUID.randomUUID().toString());
31      kafkaSpoutConfig.bufferSizeBytes = 1024 * 1024 * 4;
32      kafkaSpoutConfig.fetchSizeBytes = 1024 * 1024 * 4;
33      kafkaSpoutConfig.forceFromStart = true;
34      kafkaSpoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
35
36      TopologyBuilder builder = new TopologyBuilder();
37      builder.setSpout("kafka-spout", new KafkaSpout(kafkaSpoutCon-fig));
38      builder.setBolt("word-spitter", new SplitBolt()).shuffleGroup-ing("kafka-spout");
39      builder.setBolt("word-counter", new CountBolt()).shuffleGroup-ing("word-spitter");
40         
41      LocalCluster cluster = new LocalCluster();
42      cluster.submitTopology("KafkaStormSample", config, builder.create-Topology());
43
44      Thread.sleep(10000);
45      
46      cluster.shutdown();
47   }
48}

在移动编译之前，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的新主题，并提供一些示例消息，如下所示：

1hello
2kafka
3storm
4spark
5test message
6another test message

现在使用以下命令执行应用程序 -

java -cp“/path/to/Kafka/apache-storm-0.9.5/lib/*”：。KafkaStormSample

本应用程序的示例输出如下：

1storm : 1
2test : 2
3spark : 1
4another : 1
5kafka : 1
6hello : 1
7message : 2

Apache Kafka教程之与Storm集成

Apache Kafka - 与Storm集成

关于Storm

与Storm融合

Bolt Creation

提交到Topology

评论区

相关推荐

Apache Kafka

Apache Kafka教程 之 与Storm集成

Apache Kafka - 与Storm集成

关于Storm

与Storm融合

Bolt Creation

提交到Topology

评论区

相关推荐

Apache Kafka教程之与Storm集成