Apache Kafka-Storm과 통합

이 장에서는 Kafka를 Apache Storm과 통합하는 방법을 배웁니다.

Storm 정보

Storm은 원래 Nathan Marz와 BackType 팀이 만들었습니다. 단기간에 Apache Storm은 방대한 양의 데이터를 처리 할 수있는 분산 실시간 처리 시스템의 표준이되었습니다. Storm은 매우 빠르며 벤치 마크에 따르면 노드 당 초당 처리되는 튜플이 백만 개가 넘습니다. Apache Storm은 지속적으로 실행되어 구성된 소스 (Spouts)의 데이터를 사용하고 데이터를 처리 파이프 라인 (Bolts)으로 전달합니다. 결합 된 주둥이와 볼트는 토폴로지를 만듭니다.

Storm과 통합

Kafka와 Storm은 자연스럽게 서로를 보완하며 강력한 협력을 통해 빠르게 이동하는 빅 데이터에 대한 실시간 스트리밍 분석이 가능합니다. Kafka와 Storm 통합은 개발자가 Storm 토폴로지에서 데이터 스트림을 더 쉽게 수집하고 게시 할 수 있도록합니다.

개념적 흐름

주둥이는 개울의 원천입니다. 예를 들어, 스파우트는 Kafka 주제에서 튜플을 읽고 스트림으로 방출 할 수 있습니다. 볼트는 입력 스트림을 소비하고 처리하고 새로운 스트림을 방출 할 수 있습니다. 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)

Hosts − BrokerHost는 ZkHosts / StaticHosts가 될 수 있습니다.

Topic − 주제 이름.

SpoutConfig API

Spoutconfig는 추가 ZooKeeper 정보를 지원하는 KafkaConfig의 확장입니다.

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

• Hosts − BrokerHosts는 BrokerHosts 인터페이스의 모든 구현이 될 수 있습니다.

• Topic − 주제 이름.

• zkRoot − ZooKeeper 루트 경로.

• id −주둥이는 Zookeeper에서 소비 한 오프셋 상태를 저장합니다. ID는 스파우트를 고유하게 식별해야합니다.

SchemeAsMultiScheme

SchemeAsMultiScheme은 Kafka에서 소비 된 ByteBuffer가 스톰 튜플로 변환되는 방식을 지시하는 인터페이스입니다. MultiScheme에서 파생되며 Scheme 클래스의 구현을 허용합니다. Scheme 클래스의 많은 구현이 있으며 이러한 구현 중 하나는 바이트를 간단한 문자열로 구문 분석하는 StringScheme입니다. 또한 출력 필드의 이름을 제어합니다. 서명은 다음과 같이 정의됩니다.

public SchemeAsMultiScheme(Scheme scheme)

• Scheme − kafka에서 소비 된 바이트 버퍼.

KafkaSpout API

KafkaSpout는 Storm과 통합되는 스파우트 구현입니다. kafka 토픽에서 메시지를 가져와 스톰 생태계에 튜플로 내 보냅니다. 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는 튜플을 입력으로 사용하고 튜플을 처리하고 새 튜플을 출력으로 생성하는 구성 요소입니다. Bolts는 IRichBolt 인터페이스를 구현합니다. 이 프로그램에서는 두 개의 볼트 클래스 WordSplitter-Bolt 및 WordCounterBolt가 작업을 수행하는 데 사용됩니다.

IRichBolt 인터페이스에는 다음과 같은 방법이 있습니다.

• Prepare− 볼트에 실행할 환경을 제공합니다. 실행자는이 메서드를 실행하여 스파우트를 초기화합니다.

• Execute − 단일 튜플 입력을 처리합니다.

• Cleanup − 볼트가 셧다운 될 때 호출됩니다.

• 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 클래스에는 스파우트 (setSpout)를 설정하고 볼트 (setBolt)를 설정하는 메서드가 있습니다. 마지막으로 TopologyBuilder에는 to-pology를 생성하는 createTopology가 있습니다. shuffleGrouping 및 fieldsGrouping 메소드는 주둥이와 볼트에 대한 스트림 그룹화를 설정하는 데 도움이됩니다.

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 통합에는 큐레이터 ZooKeeper 클라이언트 자바 라이브러리가 필요합니다. 큐레이터 버전 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