Apache Storm #3 (Stream windows API: Trident)

Apache Storm

Stream windows API (Trident)

Trident의 Stream API중 window()가 있다. 이는 Streaming의 특징인 windowing 을 지원하는 메소드이다.

간단히 설명해 windowing은 batch 처리를 할때 Tuple의 양을 일정 기준에 따라 정해서 처리하는 것이다.

Storm 공식 문서에는 아래와 같이 설명하고 있다.

Trident streams can process tuples in batches which are of the same window and emit aggregated result to the next operation. There are two kinds of windowing supported which are based on processing time or tuples count: 1. Tumbling window 2. Sliding window

위 문구에서 설명한대로 Tumbling 과 Sliding 두 종류가 있고 그안에서 시간과 튜플 기반의 프로세싱으로 나뉜다. 

1. Tumbling Window

Tumbling windows는 중복 튜플을 처리하지 않는다. 튜플은 무조건 하나의 윈도우 안에서 처리되는 개념이다.

1.1 Tumbling window based on Tuple Count

튜플 갯수 기반의 Tumbling은 단순하다. 윈도우 크기만큼 유니크하게 튜플을 처리하는 구조이다.

아래 결과화면은 알파벳을 A부터 P까지 순차적으로 받는 FixedSpout에서 나오는 데이터를 Tuple Count 기반의 Tumbling window를 사용해서 문자를 붙이는 예제이다.

순서에 차이는 있지만 ABCDEF 의 알파벳 순서의 앞 5개가 처음 윈도우에서 처리되고 두번째 윈도우에서는 그 다음 알파벳 5개인 GHIJKL이 처리됬다. 한 튜플은 하나의 윈도우에서만 고정적으로 처리된다.

* Tumbling Window based on tuple count (Windows Size: 5)

2.2 Tumbling window based on Processing Time

처리 시간 기반의 Tumbling은 설정한 기간안에 들어오는 튜플들을 처리하고, 다음 주기에 들어온 튜플을 다시 처리하는 방식으로 처리된다.

예를 들어, 3초의 인터벌로 설정했다면 1:00 ~ 1:03에 들어온 튜플들을 모두 처리하고 다시 1:03 ~ 1:06에 들어온 데이터를 처리하는 구조이다.

100ms 단위로 window 크기를 설정한다. 아래 예제에서는 평균적으로 6개의 튜플이 100ms로 처리가 된다. 해당 시간에 들어온 튜플은 튜플 갯수 기반의 방식과 마찬가지로 오직 하나의 윈도우에서만 처리가 된다. 그리고 때에 따라서 5번째줄의 출력같이 3개만 처리될 경우도 있다.

* Tumbling Window based on processing time (Inteval time: 100 ms)

아래는 Tumbling Window를 사용하는 메소드이다.

//Returns a stream of tuples which are aggregated results of a tumbling window with every {@code windowCount} of tuples. public Stream tumblingWindow(int windowCount, WindowsStoreFactory windowStoreFactory, Fields inputFields, Aggregator aggregator, Fields functionFields); //Returns a stream of tuples which are aggregated results of a window that tumbles at duration of {@code windowDuration} public Stream tumblingWindow(BaseWindowedBolt.Duration windowDuration, WindowsStoreFactory windowStoreFactory, Fields inputFields, Aggregator aggregator, Fields functionFields);

2. Sliding Window

Sliding windows는 튜플을 중복으러 처리한다. 예를 들어 지금 시간 기준으로 10분간의 데이터에서 평균을 구하는 연산이 있다면 동일한 튜플을 10분이라는 기간안에서 중복 처리가 되어야 한다.

2.1 Sliding window based on Tuple Count

튜플 갯수 기반의 Sliding 은 윈도우 갯수와 Sliding 갯수를 설정할 수 있다.

아래 예제는 윈도우 사이즈 10에 슬라이딩 사이즈는 3이다. 3개의 튜플씩 윈도우에 들어간다. 그리고 10개가 넘을 경우 가장 먼저 들어온 3개의 튜플이 제거되고 새로 3개의 튜플이 추가 되어 하나의 윈도우를 만든다.

* Sliding Window based on tuple count (Windows Size: 10, Sliding size: 3)

2.2 Sliding window based on Processing Time

프로세싱 타임 기반의 슬라이딩 윈도우는 윈도우의 인터벌과 슬라이딩의 인터벌을 각각 지정해야 한다.

아래 예제는 윈도우 인터벌을 300ms, 슬라이딩 인터벌을 100ms로 지정한 예제이다. 100ms에 프로세싱 되는 퓨틀이 약 6개이고 윈도우 인터벨 300ms 에 처리되는 튜플이 총 18개 정도가 된다. 따라서 프로세싱 타임에 일정한 튜플의 양이 처리된다는 가정으로 튜플 카운트의 18 / 6 으로 설정한 세팅과 유사하게 처리 될 것이다.

* Sliding Window based on Processing time 

  (Windows Interval: 300ms, Sliding Interval: 100ms)

아래는 Sliding Window를 사용하는 메소드이다.

// Returns a stream of tuples which are aggregated results of a sliding window with every {@code windowCount} // of tuples and slides the window after {@code slideCount}. public Stream slidingWindow(int windowCount, int slideCount, WindowsStoreFactory windowStoreFactory , Fields inputFields, Aggregator aggregator, Fields functionFields); //Returns a stream of tuples which are aggregated results of a window which slides at duration of {@code slidingInterval} // and completes a window at {@code windowDuration} public Stream slidingWindow(BaseWindowedBolt.Duration windowDuration, BaseWindowedBolt.Duration slidingInterval , WindowsStoreFactory windowStoreFactory, Fields inputFields, Aggregator aggregator, Fields functionFields);

3. 예제 소스 코드

아래 예제코드는 각 윈도우 타입에 따라 결과 코드에 대한 코드이다. Spout 데이터와 연산방법은 동일하게 설정하고 함수도 window()를 사용했다. 거기에 첫번째 파라메터인 WindowConfig를 다르게 설정하여 위 4 예제에 대한 결과를 얻을 수 있다.

package com.minsub.storm.example.trident.api; import org.apache.storm.Config; import org.apache.storm.LocalCluster; import org.apache.storm.generated.StormTopology; import org.apache.storm.topology.base.BaseWindowedBolt; import org.apache.storm.trident.TridentTopology; import org.apache.storm.trident.operation.BaseAggregator; import org.apache.storm.trident.operation.TridentCollector; import org.apache.storm.trident.testing.FixedBatchSpout; import org.apache.storm.trident.tuple.TridentTuple; import org.apache.storm.trident.windowing.InMemoryWindowsStoreFactory; import org.apache.storm.trident.windowing.config.*; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Values; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.TimeUnit; public class WindowsTopology { public static StormTopology buildTopology() throws Exception { TridentTopology topology = new TridentTopology(); // Spout FixedBatchSpout spout = new FixedBatchSpout(new Fields("char"), 3 , new Values("A"), new Values("B"), new Values("E"), new Values("F"), new Values("C"), new Values("D") , new Values("G"), new Values("H"), new Values("I"), new Values("J"), new Values("K"), new Values("L") , new Values("M"), new Values("N"), new Values("O"), new Values("P"), new Values("Q"), new Values("R")); spout.setCycle(true); WindowConfig windowConfig; windowConfig = TumblingCountWindow.of(3); //windowConfig = SlidingCountWindow.of(10, 3); //windowConfig = SlidingDurationWindow.of(new BaseWindowedBolt.Duration(300, TimeUnit.MILLISECONDS), new BaseWindowedBolt.Duration(100, TimeUnit.MILLISECONDS)); //windowConfig = TumblingDurationWindow.of(new BaseWindowedBolt.Duration(100, TimeUnit.MILLISECONDS)); topology.newStream("char", spout) .window(windowConfig, new InMemoryWindowsStoreFactory(), new Fields("char"), new BaseAggregator() { private StringBuilder sb; @Override public Object init(Object o, TridentCollector tridentCollector) { this.sb = new StringBuilder(); return null; } @Override public void complete(Object o, TridentCollector collector) { collector.emit(new Values(this.sb.toString())); } @Override public void aggregate(Object o, TridentTuple tuple, TridentCollector collector) { sb.append(tuple.getString(0)); } }, new Fields("merge")) .peek(tuple -> System.out.println(tuple)); return topology.build(); } public static void main(String[] args) throws Exception { Config conf = new Config(); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("windows-topology", conf, buildTopology()); Thread.sleep(1000 * 20); cluster.killTopology("windows-topology"); cluster.shutdown(); } }

끝