Apache Kafka-Stormとの統合

この章では、KafkaをApache Stormと統合する方法を学びます。

ストームについて

Stormは元々Nathan MarzとBackTypeのチームによって作成されました。短時間で、Apache Stormは膨大な量のデータを処理できる分散リアルタイム処理システムの標準になりました。 Stormは非常に高速であり、ベンチマークでは、ノードごとに1秒あたり100万を超えるタプルが処理されました。 Apache Stormは継続的に実行され、構成されたソース（Spouts）からデータを消費し、データを処理パイプライン（Bolts）に渡します。組み合わせて、スパウトとボルトがトポロジを作成します。

Stormとの統合

KafkaとStormは自然にお互いを補完し、強力な協力により、高速で移動するビッグデータのリアルタイムストリーミング分析が可能になります。 KafkaとStormの統合は、開発者がStormトポロジからデータストリームを簡単に取り込み、公開できるようにすることです。

概念フロー

注ぎ口は、ストリームのソースです。たとえば、スパウトは、Kafkaトピックからタプルを読み取り、ストリームとして放出する場合があります。ボルトは入力ストリームを消費し、処理し、場合によっては新しいストリームを放出します。 Boltsは、関数の実行、タプルのフィルタリング、ストリーミング集約、ストリーミング結合、データベースとの対話など、あらゆることを実行できます。 Stormトポロジの各ノードは並行して実行されます。トポロジは、終了するまで無期限に実行されます。 Stormは、失敗したタスクを自動的に再割り当てします。さらに、Stormは、マシンがダウンしてメッセージがドロップされた場合でも、データの損失がないことを保証します。

Kafka-Storm統合APIの詳細を見ていきましょう。 KafkaをStormに統合するには、3つの主要なクラスがあります。彼らは次のとおりです-

BrokerHosts-ZkHostsおよびStaticHosts

BrokerHostsはインターフェースであり、ZkHostsとStaticHostsはその2つの主要な実装です。 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)

SpoutConfig API

Spoutconfigは、追加のZooKeeper情報をサポートするKafkaConfigの拡張機能です。

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

Hosts -BrokerHostsは、BrokerHostsインターフェイスの任意の実装にすることができます
トピック-トピック名。
zkRoot -ZooKeeperのルートパス。
* id-*注ぎ口は、Zookeeperで消費されたオフセットの状態を保存します。 idは、スパウトを一意に識別する必要があります。

SchemeAsMultiScheme

SchemeAsMultiSchemeは、Kafkaから消費されたByteBufferがストームタプルに変換される方法を指示するインターフェイスです。 MultiSchemeから派生し、Schemeクラスの実装を受け入れます。 Schemeクラスには多くの実装があり、そのような実装の1つにStringSchemeがあります。これは、バイトを単純な文字列として解析します。また、出力フィールドの命名も制御します。署名は次のように定義されます。

public SchemeAsMultiScheme(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は、タプルを入力として受け取り、タプルを処理し、出力として新しいタプルを生成するコンポーネントです。ボルトはIRichBoltインターフェイスを実装します。このプログラムでは、2つのボルトクラス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<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()&plus;" : " &plus; 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ライブラリが必要です。キュレーターバージョン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-integration-storm