Kafka 与Storm集成



  • Apache Storm集成。

    在本章中,我们将学习如何将Kafka与Apache Storm集成。
  • 关于Storm

    Storm最初是由内森·马兹(Nathan Marz)和BackType团队创建的。在短时间内,Apache Storm成为分布式实时处理系统的标准,该系统使您可以处理大量数据。Storm的速度非常快,基准测试表明它每秒可处理每个节点超过一百万个元组。Apache Storm持续运行,使用来自已配置源(Spouts)的数据,并将数据向下传递到处理管道(Bolts)。喷头和螺栓结合在一起构成一个拓扑。
  • 与Storm整合

    Kafka和Storm自然互补,它们的强大合作可实现实时流分析,以快速移动大数据。Kafka和Storm的集成将使开发人员更轻松地从Storm拓扑中提取和发布数据流。
    概念流
    壶嘴是溪流的源头。例如,喷口可能会从Kafka主题中读取元组并将其作为流发出。螺栓消耗输入流,处理并可能发出新流。螺栓可以执行任何操作,包括运行函数,过滤元组,执行流聚合,流连接,与数据库对话等等。Storm拓扑中的每个节点都并行执行。拓扑无限期运行,直到您终止它。Storm将自动重新分配所有失败的任务。此外,Storm保证即使机器宕机和消息丢失也不会丢失数据。
    让我们详细了解Kafka-Storm集成API。有三个主要的类将Kafka与Storm集成。它们如下-
    BrokerHosts-ZkHosts和StaticHosts
    BrokerHosts是一个接口,ZkHostsStaticHosts是它的两个主要实现。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集群的配置设置。KafkaConfig的签名定义如下
    
    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- 喷口存储其在Zookeeper中消耗的偏移量的状态。该ID应该唯一标识您的壶嘴。
    SchemeAsMultiScheme
    SchemeAsMultiScheme是一个接口,指示从Kafka消耗的ByteBuffer如何转换为Storm元组。它从MultiScheme派生并接受Scheme类的实现。Scheme类有很多实现,StringScheme是这样的实现,它将字节解析为简单字符串。它还控制输出字段的命名。签名定义如下。
    
    public SchemeAsMultiScheme(Scheme scheme)
    
    
    • scheme - 从Kafka消耗的字节缓冲区。
    KafkaSpout API
    KafkaSpout是我们的spout实施,它将与Storm集成。它从kafka主题中获取消息,并将其作为元组发送到Storm生态系统中。KafkaSpout从SpoutConfig获取其配置详细信息。
    以下是创建简单的KafkaSpout的示例代码。
    
    // 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是一个将元组作为输入,处理该元组并生成新的元组作为输出的组件。螺栓(Bolt)将实现IRichBolt接口。在此程序中,两个螺栓类WordSplitterBolt和WordCounterBolt用于执行操作。
    IRichBolt接口具有以下方法-
    • Prepare - 为螺栓提供执行环境。执行者将运行此方法初始化喷口。
    • Execute - 处理单个输入元组。
    • Cleanup - 当螺栓即将关闭时调用。
    • clarifyOutputFields - 声明元组的输出模式。
    让我们创建SplitBolt.java和CountBolt.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类具有设置喷口(setSpout)和设置螺栓(setBolt)的方法。最后,TopologyBuilder具有createTopology来创建拓扑。shuffleGrouping和fieldsGrouping方法有助于为喷口和螺栓设置流分组。
    本地群集 - 出于开发目的,我们可以使用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集成需要管理者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(在上一章中进行了说明),创建一个名为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