❮ Apache Kafka 消费者组示例 Apache Kafka 与 Spark 集成 ❯

Apache Kafka - 与 Storm 集成

更新于 2024/9/16 10:48:00

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

关于 Storm

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

与 Storm 集成

Kafka 和 Storm 自然相得益彰，强大的合作为快速移动的大数据提供实时流式分析。 Kafka 和 Storm 的集成是为了让开发人员更容易从 Storm 拓扑中摄取和发布数据流。

概念流程

spout 是流的来源。例如，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 是 ZooKeeper 路径，用于维护 Kafka 代理详细信息。

KafkaConfig API

此 API 用于定义 Kafka 集群的配置设置。 Kafka Config 的签名定义如下

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

SchemeAsMultiScheme

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

public SchemeAsMultiScheme(Scheme scheme)

Scheme − 从 kafka 消耗的字节缓冲区。

KafkaSpout API

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

下面是创建简单 Kafka spout 的示例代码。

// 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 是一个将元组作为输入，处理元组，并产生新元组作为输出的组件。 Bolts 将实现 IRichBolt 接口。在这个程序中，两个 bolt 类 WordSplitter-Bolt 和 WordCounterBolt 用于执行操作。

IRichBolt 接口有以下方法 −

Prepare − 为 bolt 提供执行环境。执行程序将运行此方法来初始化 spout。
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;
   }
}

提交拓扑

Storm 拓扑基本上是一个 Thrift 结构。 TopologyBuilder 类提供了简单易用的方法来创建复杂的拓扑。 TopologyBuilder 类具有设置 spout (setSpout) 和设置 Bolt (setBolt) 的方法。最后，TopologyBuilder 有 createTopology 来创建拓扑。 shuffleGrouping 和 fieldsGrouping 方法有助于为 spout 和 bolts 设置流分组。

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 客户端 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

❮ Apache Kafka 消费者组示例 Apache Kafka 与 Spark 集成 ❯

Apache Kafka 教程

Apache Kafka 资源