Apache Storm-작업 예제

Apache Storm의 핵심 기술 세부 사항을 살펴 보았으므로 이제 몇 가지 간단한 시나리오를 코딩 할 차례입니다.

시나리오 – 모바일 통화 로그 분석기

모바일 통화 및 통화 시간은 Apache Storm에 대한 입력으로 제공되며 Storm은 동일한 발신자와 수신자 간의 통화 및 총 통화 수를 처리하고 그룹화합니다.

주둥이 만들기

Spout는 데이터 생성에 사용되는 구성 요소입니다. 기본적으로 스파우트는 IRichSpout 인터페이스를 구현합니다. “IRichSpout”인터페이스에는 다음과 같은 중요한 방법이 있습니다.

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

  • nextTuple − 생성 된 데이터를 수집기를 통해 방출합니다.

  • close −이 메서드는 스파우트가 종료 될 때 호출됩니다.

  • declareOutputFields − 튜플의 출력 스키마를 선언합니다.

  • ack − 특정 튜플이 처리되었음을 확인

  • fail − 특정 튜플이 처리되지 않고 재 처리되지 않도록 지정합니다.

열다

의 서명 open 방법은 다음과 같습니다-

open(Map conf, TopologyContext context, SpoutOutputCollector collector)
  • conf −이 스파우트에 대한 스톰 구성을 제공합니다.

  • context − 토폴로지 내 스파우트 위치, 작업 ID, 입력 및 출력 정보에 대한 완전한 정보를 제공합니다.

  • collector − 볼트에 의해 처리 될 튜플을 방출 할 수 있습니다.

nextTuple

의 서명 nextTuple 방법은 다음과 같습니다-

nextTuple()

nextTuple ()은 ack () 및 fail () 메서드와 동일한 루프에서 주기적으로 호출됩니다. 수행 할 작업이 없을 때 스레드의 제어를 해제해야 다른 메서드가 호출 될 수 있습니다. 따라서 nextTuple의 첫 번째 줄은 처리가 완료되었는지 확인합니다. 그렇다면 반환하기 전에 프로세서의 부하를 줄이기 위해 최소 1 밀리 초 동안 절전 모드를 유지해야합니다.

닫기

의 서명 close 방법은 다음과 같습니다-

close()

declareOutputFields

의 서명 declareOutputFields 방법은 다음과 같습니다-

declareOutputFields(OutputFieldsDeclarer declarer)

declarer − 출력 스트림 ID, 출력 필드 등을 선언하는 데 사용됩니다.

이 메서드는 튜플의 출력 스키마를 지정하는 데 사용됩니다.

확인

의 서명 ack 방법은 다음과 같습니다-

ack(Object msgId)

이 메서드는 특정 튜플이 처리되었음을 확인합니다.

불합격

의 서명 nextTuple 방법은 다음과 같습니다-

ack(Object msgId)

이 메서드는 특정 튜플이 완전히 처리되지 않았 음을 알려줍니다. Storm은 특정 튜플을 다시 처리합니다.

FakeCallLogReaderSpout

이 시나리오에서는 통화 기록 세부 정보를 수집해야합니다. 통화 기록의 정보가 포함됩니다.

  • 발신자 번호
  • 수신자 번호
  • duration

실시간 통화 기록 정보가 없기 때문에 가짜 통화 기록을 생성합니다. 가짜 정보는 Random 클래스를 사용하여 생성됩니다. 전체 프로그램 코드는 다음과 같습니다.

코딩-FakeCallLogReaderSpout.java

import java.util.*;
//import storm tuple packages
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

//import Spout interface packages
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;

//Create a class FakeLogReaderSpout which implement IRichSpout interface 
   to access functionalities
	
public class FakeCallLogReaderSpout implements IRichSpout {
   //Create instance for SpoutOutputCollector which passes tuples to bolt.
   private SpoutOutputCollector collector;
   private boolean completed = false;
	
   //Create instance for TopologyContext which contains topology data.
   private TopologyContext context;
	
   //Create instance for Random class.
   private Random randomGenerator = new Random();
   private Integer idx = 0;

   @Override
   public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
      this.context = context;
      this.collector = collector;
   }

   @Override
   public void nextTuple() {
      if(this.idx <= 1000) {
         List<String> mobileNumbers = new ArrayList<String>();
         mobileNumbers.add("1234123401");
         mobileNumbers.add("1234123402");
         mobileNumbers.add("1234123403");
         mobileNumbers.add("1234123404");

         Integer localIdx = 0;
         while(localIdx++ < 100 && this.idx++ < 1000) {
            String fromMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
            String toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
				
            while(fromMobileNumber == toMobileNumber) {
               toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
            }
				
            Integer duration = randomGenerator.nextInt(60);
            this.collector.emit(new Values(fromMobileNumber, toMobileNumber, duration));
         }
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("from", "to", "duration"));
   }

   //Override all the interface methods
   @Override
   public void close() {}

   public boolean isDistributed() {
      return false;
   }

   @Override
   public void activate() {}

   @Override 
   public void deactivate() {}

   @Override
   public void ack(Object msgId) {}

   @Override
   public void fail(Object msgId) {}

   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
}

볼트 생성

Bolt는 튜플을 입력으로 사용하고 튜플을 처리하고 새 튜플을 출력으로 생성하는 구성 요소입니다. 볼트는IRichBolt상호 작용. 이 프로그램에서 두 개의 볼트 클래스CallLogCreatorBoltCallLogCounterBolt 작업을 수행하는 데 사용됩니다.

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

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

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

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

  • declareOutputFields − 튜플의 출력 스키마를 선언합니다.

준비

의 서명 prepare 방법은 다음과 같습니다-

prepare(Map conf, TopologyContext context, OutputCollector collector)
  • conf −이 볼트에 대한 Storm 구성을 제공합니다.

  • context − 토폴로지 내 볼트 위치, 작업 ID, 입력 및 출력 정보 등에 대한 완전한 정보를 제공합니다.

  • collector − 처리 된 튜플을 방출 할 수 있습니다.

실행하다

의 서명 execute 방법은 다음과 같습니다-

execute(Tuple tuple)

여기 tuple 처리 할 입력 튜플입니다.

그만큼 execute메서드는 한 번에 하나의 튜플을 처리합니다. 튜플 데이터는 Tuple 클래스의 getValue 메서드를 통해 액세스 할 수 있습니다. 입력 튜플을 즉시 처리 할 필요는 없습니다. 여러 튜플을 처리하고 단일 출력 튜플로 출력 할 수 있습니다. 처리 된 튜플은 OutputCollector 클래스를 사용하여 내보낼 수 있습니다.

대청소

의 서명 cleanup 방법은 다음과 같습니다-

cleanup()

declareOutputFields

의 서명 declareOutputFields 방법은 다음과 같습니다-

declareOutputFields(OutputFieldsDeclarer declarer)

여기에 매개 변수 declarer 출력 스트림 ID, 출력 필드 등을 선언하는 데 사용됩니다.

이 메서드는 튜플의 출력 스키마를 지정하는 데 사용됩니다.

콜 로그 Creator Bolt

콜 로그 생성자 볼트는 콜 로그 튜플을 수신합니다. 통화 기록 튜플에는 발신자 번호, 수신자 번호 및 통화 기간이 있습니다. 이 볼트는 단순히 발신자 번호와 수신자 번호를 결합하여 새로운 값을 생성합니다. 새 값의 형식은 "발신자 번호 – 수신자 번호"이며 새 필드 "통화"로 이름이 지정됩니다. 전체 코드는 다음과 같습니다.

코딩-CallLogCreatorBolt.java

//import util packages
import java.util.HashMap;
import java.util.Map;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;

//import Storm IRichBolt package
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

//Create a class CallLogCreatorBolt which implement IRichBolt interface
public class CallLogCreatorBolt implements IRichBolt {
   //Create instance for OutputCollector which collects and emits tuples to produce output
   private OutputCollector collector;

   @Override
   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
      this.collector = collector;
   }

   @Override
   public void execute(Tuple tuple) {
      String from = tuple.getString(0);
      String to = tuple.getString(1);
      Integer duration = tuple.getInteger(2);
      collector.emit(new Values(from + " - " + to, duration));
   }

   @Override
   public void cleanup() {}

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("call", "duration"));
   }
	
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
}

콜 로그 카운터 볼트

콜 로그 카운터 볼트는 콜과 그 기간을 튜플로받습니다. 이 볼트는 prepare 메소드에서 사전 (Map) 객체를 초기화합니다. 에execute메소드는 튜플을 확인하고 튜플의 모든 새로운 "호출"값에 대해 사전 객체에 새 항목을 만들고 사전 객체에 값 1을 설정합니다. 사전에서 이미 사용 가능한 항목의 경우 값을 증가시킵니다. 간단히 말해서,이 볼트는 사전 개체에 호출 및 해당 개수를 저장합니다. 사전에 호출 및 해당 개수를 저장하는 대신 데이터 소스에 저장할 수도 있습니다. 전체 프로그램 코드는 다음과 같습니다.

코딩-CallLogCounterBolt.java

import java.util.HashMap;
import java.util.Map;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

public class CallLogCounterBolt implements IRichBolt {
   Map<String, Integer> counterMap;
   private OutputCollector collector;

   @Override
   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
      this.counterMap = new HashMap<String, Integer>();
      this.collector = collector;
   }

   @Override
   public void execute(Tuple tuple) {
      String call = tuple.getString(0);
      Integer duration = tuple.getInteger(1);
		
      if(!counterMap.containsKey(call)){
         counterMap.put(call, 1);
      }else{
         Integer c = counterMap.get(call) + 1;
         counterMap.put(call, c);
      }
		
      collector.ack(tuple);
   }

   @Override
   public void cleanup() {
      for(Map.Entry<String, Integer> entry:counterMap.entrySet()){
         System.out.println(entry.getKey()+" : " + entry.getValue());
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("call"));
   }
	
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
	
}

토폴로지 생성

Storm 토폴로지는 기본적으로 Thrift 구조입니다. TopologyBuilder 클래스는 복잡한 토폴로지를 만드는 간단하고 쉬운 방법을 제공합니다. TopologyBuilder 클래스에는 스파우트를 설정하는 메서드가 있습니다.(setSpout) 볼트를 설정하려면 (setBolt). 마지막으로 TopologyBuilder에는 토폴로지를 생성하는 createTopology가 있습니다. 다음 코드 조각을 사용하여 토폴로지를 만듭니다.

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());

builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt())
   .shuffleGrouping("call-log-reader-spout");

builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt())
   .fieldsGrouping("call-log-creator-bolt", new Fields("call"));

shuffleGroupingfieldsGrouping 메서드는 주둥이와 볼트에 대한 스트림 그룹화를 설정하는 데 도움이됩니다.

로컬 클러스터

개발 목적으로 "LocalCluster"개체를 사용하여 로컬 클러스터를 만든 다음 "LocalCluster"클래스의 "submitTopology"메서드를 사용하여 토폴로지를 제출할 수 있습니다. "submitTopology"에 대한 인수 중 하나는 "Config"클래스의 인스턴스입니다. "Config"클래스는 토폴로지를 제출하기 전에 구성 옵션을 설정하는 데 사용됩니다. 이 구성 옵션은 런타임에 클러스터 구성과 병합되고 준비 방법을 사용하여 모든 작업 (spout 및 bolt)으로 전송됩니다. 토폴로지가 클러스터에 제출되면 클러스터가 제출 된 토폴로지를 계산할 때까지 10 초간 기다린 다음 "LocalCluster"의 "종료"방법을 사용하여 클러스터를 종료합니다. 전체 프로그램 코드는 다음과 같습니다.

코딩-LogAnalyserStorm.java

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

//import storm configuration packages
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

//Create main class LogAnalyserStorm submit topology.
public class LogAnalyserStorm {
   public static void main(String[] args) throws Exception{
      //Create Config instance for cluster configuration
      Config config = new Config();
      config.setDebug(true);
		
      //
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());

      builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt())
         .shuffleGrouping("call-log-reader-spout");

      builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt())
         .fieldsGrouping("call-log-creator-bolt", new Fields("call"));
			
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("LogAnalyserStorm", config, builder.createTopology());
      Thread.sleep(10000);
		
      //Stop the topology
		
      cluster.shutdown();
   }
}

응용 프로그램 빌드 및 실행

완전한 애플리케이션에는 4 개의 Java 코드가 있습니다. 그들은-

  • FakeCallLogReaderSpout.java
  • CallLogCreaterBolt.java
  • CallLogCounterBolt.java
  • LogAnalyerStorm.java

응용 프로그램은 다음 명령을 사용하여 구축 할 수 있습니다-

javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*” *.java

응용 프로그램은 다음 명령을 사용하여 실행할 수 있습니다-

java -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:. LogAnalyserStorm

산출

응용 프로그램이 시작되면 클러스터 시작 프로세스, 스파우트 및 볼트 처리, 마지막으로 클러스터 종료 프로세스에 대한 전체 세부 정보가 출력됩니다. "CallLogCounterBolt"에서 호출 및 카운트 세부 정보를 인쇄했습니다. 이 정보는 다음과 같이 콘솔에 표시됩니다.

1234123402 - 1234123401 : 78
1234123402 - 1234123404 : 88
1234123402 - 1234123403 : 105
1234123401 - 1234123404 : 74
1234123401 - 1234123403 : 81
1234123401 - 1234123402 : 81
1234123403 - 1234123404 : 86
1234123404 - 1234123401 : 63
1234123404 - 1234123402 : 82
1234123403 - 1234123402 : 83
1234123404 - 1234123403 : 86
1234123403 - 1234123401 : 93

비 JVM 언어

Storm 토폴로지는 Thrift 인터페이스로 구현되어 모든 언어로 토폴로지를 쉽게 제출할 수 있습니다. Storm은 Ruby, Python 및 기타 여러 언어를 지원합니다. 파이썬 바인딩을 살펴 보겠습니다.

파이썬 바인딩

Python은 범용 해석, 대화 형, 객체 지향 및 고급 프로그래밍 언어입니다. Storm은 Python을 지원하여 토폴로지를 구현합니다. Python은 방출, 고정, 접근 및 로깅 작업을 지원합니다.

아시다시피 볼트는 모든 언어로 정의 할 수 있습니다. 다른 언어로 작성된 Bolt는 하위 프로세스로 실행되고 Storm은 stdin / stdout을 통해 JSON 메시지를 사용하여 해당 하위 프로세스와 통신합니다. 먼저 파이썬 바인딩을 지원하는 샘플 볼트 WordCount를 가져옵니다.

public static class WordCount implements IRichBolt {
   public WordSplit() {
      super("python", "splitword.py");
   }
	
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("word"));
   }
}

여기 수업 WordCount 구현 IRichBolt인터페이스 및 파이썬 구현에 지정된 수퍼 메소드 인수 "splitword.py"로 실행됩니다. 이제 "splitword.py"라는 파이썬 구현을 만듭니다.

import storm
   class WordCountBolt(storm.BasicBolt):
      def process(self, tup):
         words = tup.values[0].split(" ")
         for word in words:
         storm.emit([word])
WordCountBolt().run()

이것은 주어진 문장에서 단어를 세는 Python의 샘플 구현입니다. 마찬가지로 다른 지원 언어와도 바인딩 할 수 있습니다.