开发手册 欢迎您！
软件开发者资料库

Apache Kafka - 与Storm集成

Apache Kafka与Storm集成 - 从简介，基础，集群架构，工作流程，安装步骤，基本操作，简单生产者示例，消费者组示例，与Storm集成，与Spark集成，实时应用程序（Twitter），学习Apache kafka，工具，应用程序。

在本章中，我们将学习如何将Kafka与Apache Storm集成.

关于Storm

Storm最初是由Nathan Marz创建的BackType团队.在短时间内，Apache Storm成为分布式实时处理系统的标准，允许您处理大量数据.风暴速度非常快，每个节点每秒处理的元组数超过一百万个. Apache Storm持续运行，消耗来自配置源(Spouts)的数据，并将数据传递到处理管道(Bolts).联合，Spouts和Bolts构成拓扑.

与Storm集成

Kafka和Storm自然相互补充，他们强大的合作使真实用于快速移动的大数据的时间流分析. Kafka和Storm的集成是为了让开发人员更容易从Storm拓扑中摄取和发布数据流.

概念流

喷口是一个来源流.例如，一个spout可以从Kafka主题中读取元组并将它们作为流发出. bolt会消耗输入流，进程并可能发出新的流. 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路径.

KafkaConfig API

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

 public KafkaConfig(BrokerHosts hosts，string topic)

    主持人 :  BrokerHosts可以是ZkHosts/StaticHosts.

    主题 : 主题名称.

SpoutConfig API

Spoutconfig是KafkaConfig的扩展，支持其他ZooKeeper信息.

 public SpoutConfig(BrokerHosts hosts，string topic，string zkRoot，string id)

  • 主持人 :  BrokerHosts可以是BrokerHosts界面的任何实现

  • 主题 : 主题名称.

  • zkRoot :  ZooKeeper根路径.

  • id :   spout存储在Zookeeper中消耗的偏移量的状态. id应唯一标识您的鲸鱼喷水.

SchemeAsMultiScheme

SchemeAsMultiScheme是一个界面，指示如何从Kafka消耗的ByteBuffer变成了风暴元组.它源自MultiScheme并接受Scheme类的实现.有很多Scheme类的实现，一个这样的实现是StringScheme，它将字节解析为一个简单的字符串.它还控制输出字段的命名.签名定义如下.

public SchemeAsMultiScheme(Scheme scheme)

  • Scheme : 从kafka消耗的字节缓冲区.

KafkaSpout API

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

下面是创建简单Kafka喷口的示例代码.

//ZooKeeper连接字符串 BrokerHosts hosts = new ZkHosts(zkConnString); //创建对象 SpoutConfig spoutConfig = new SpoutConfig(hosts， topicName，"/"+ topicName UUID.randomUUID().toString ()); //将ByteBuffer转换为String.  spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme()); //将SpoutConfig分配给KafkaSpout.  KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

Bolt Creation

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

IRichBolt接口具有以下方法 :

  • 准备 : 为螺栓提供要执行的环境.执行者将运行此方法来初始化喷口.

  • 执行 : 处理单个输入元组.

  • 清理 : 当螺栓关闭时调用.

  • 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 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 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类具有设置spout(setSpout)和设置bolt(setBolt)的方法.最后，拓扑建设者创建了拓扑学来创建拓扑学. shuffleGrouping和fieldsGrouping方法有助于为spout和bolt设置流分组.

Local Cluster : 出于开发目的，我们可以使用对象创建本地群集，然后使用类的方法提交拓扑.

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库.策展人版本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(在前一章中解释)，创建一个名为并提供一些示例消息，如下所示 :

hellokafkastormsparktest messageanother test message

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

java -cp"/path/to/Kafka/apache-storm-0.9.5/lib/*": KafkaStormSample

此应用程序的示例输出指定在 :

storm : 1test : 2spark : 1another : 1kafka : 1hello : 1message : 2