Native Streaming Window APIを使ってみる
Stormのネイティブな機能として、スライディングウィンドウが追加されました。どこまでの内容までがStormネイティブとして対応しているのか、サンプルをベースに確認してみたいと思います。まずはメインパートであるSlidingWindowTopology.javaのソースを確認してみます。
public static void main(String[] args) throws Exception {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("integer", new RandomIntegerSpout(), 1);
builder.setBolt("slidingsum", new SlidingWindowSumBolt().withWindow(new Count(30), new Count(10)), 1)
.shuffleGrouping("integer");
builder.setBolt("tumblingavg", new TumblingWindowAvgBolt().withTumblingWindow(new Count(3)), 1)
.shuffleGrouping("slidingsum");
builder.setBolt("printer", new PrinterBolt(), 1).shuffleGrouping("tumblingavg");
Config conf = new Config();
conf.setDebug(true);
if (args != null && args.length > 0) {
conf.setNumWorkers(1);
StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
} else {
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("test", conf, builder.createTopology());
Utils.sleep(40000);
cluster.killTopology("test");
cluster.shutdown();
}
}
これだけ見ても、以下の3つのBoltが存在します。
- SlidingWindowSumBolt
- TumblingWindowAvgBolt
- PrinterBolt
これらが何をしているのか、またどのように動くのかを確認したいと思います。
[SlidingWindowSumBolt]
とても単純に、受信したTupleの中身を加算していることがわかります。一応ウィンドウから外れたTupleの値は減算するようにも記述されているので、スライディングウィンドウの条件を満たしていることも確認できます。
@Override
public void execute(TupleWindow inputWindow) {
List<Tuple> tuplesInWindow = inputWindow.get();
List<Tuple> newTuples = inputWindow.getNew();
List<Tuple> expiredTuples = inputWindow.getExpired();
LOG.debug("Events in current window: " + tuplesInWindow.size());
for (Tuple tuple : newTuples) {
sum += (int) tuple.getValue(0);
}
for (Tuple tuple : expiredTuples) {
sum -= (int) tuple.getValue(0);
}
collector.emit(new Values(sum));
}
[TumblingWindowAvgBolt]
設定したWindowのサイズで合計値を除算している、シンプルなつくりでした。SlidingWindowTopologyに内包されていますね。「いくつ溜まったら平均値を計算する」という引数には「3」が設定されています。
@Override
public void execute(TupleWindow inputWindow) {
int sum = 0;
List<Tuple> tuplesInWindow = inputWindow.get();
LOG.debug("Events in current window: " + tuplesInWindow.size());
if (tuplesInWindow.size() > 0) {
for (Tuple tuple : tuplesInWindow) {
sum += (int) tuple.getValue(0);
}
collector.emit(new Values(sum / tuplesInWindow.size()));
}
}
[PrinterBolt]
驚くほどシンプルですね。出力するだけ。
@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
System.out.println(tuple);
}
まとめると、以下の通りに動くと予想できます。
- ランダムに0~999の整数を、合計用のBoltに送付する。
- 合計用Boltはデータを受信時、メモリに保持しているSum値に加算していく。
- 受信したTuple数が10になったところで、平均計算用Boltに合計値を送付する。
- 1~3を、平均計算用Boltの保持Tuple数が3になったら、平均値を計算する。
- 1~4を40秒間繰り返す。
これらを基に、動かした際のログを見てみましょう。ソースコードの通り、起動後に40秒で終了するようなので、以下のコマンドを実行してしばらく待ってみます。
# bin/storm jar examples/storm-starter/storm-starter-topologies-1.0.0.jar org.apache.storm.starter.SlidingWindowTopology
[合計用Boltに対する出力ログ]
以下のような感じで合計用Boltにはログが出力されていました。
17562 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.executor - Processing received message FOR 4 TUPLE: source: integer:2, stream: default, id: {4017617819859316672=-3880300761259985782}, [899, 1461542140369, 1]
17562 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.executor - Execute done TUPLE source: integer:2, stream: default, id: {4017617819859316672=-3880300761259985782}, [899, 1461542140369, 1] TASK: 4 DELTA:
17674 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.executor - Processing received message FOR 4 TUPLE: source: integer:2, stream: default, id: {-2465951973599725012=3371764614267379312}, [888, 1461542140475, 2]
17674 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.executor - Execute done TUPLE source: integer:2, stream: default, id: {-2465951973599725012=3371764614267379312}, [888, 1461542140475, 2] TASK: 4 DELTA:
で、10個たまったところで平均計算用Boltに送付しています。
18516 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.task - Emitting: slidingsum default [6053]
18516 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.executor - TRANSFERING tuple [dest: 5 tuple: source: slidingsum:4, stream: default, id: {-1356407922762824927=-4912942846543592390, 4017617819859316672=-8950777703384980140, -7679610850762262585=5600559632140381686, -217168626838496871=6670643717321357413, -8433729321932816312=-1512990481045386819, -695350376461229364=-6915299522591467528, -8604773776820158944=2085240823323939478, 4818452273885227082=1055563511177261421, -4383830359476279213=-2430226558792731842, -2465951973599725012=-3111554498250395772}, [6053]]
:
:
19527 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.task - Emitting: slidingsum default [11182]
19527 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.executor - TRANSFERING tuple [dest: 5 tuple: source: slidingsum:4, stream: default, id: {4017617819859316672=1956648145851480265, -7679610850762262585=-8990022321548325348, -217168626838496871=8356349476653175499, 6556174365450512594=-2956830898901769282, 4973296703617984132=630324356173502412, -8433729321932816312=7530781138220324522, 8041484834072108391=-1100584463729972475, -695350376461229364=5513770714708145606, -8604773776820158944=-7212706120285088590, -4383830359476279213=5439461521018447939, -1641897178290464600=-510250118691366334, 6730277299577429107=6208397095766677293, -8115189405407159227=1214364586718890587, -1356407922762824927=3843071132908231388, 7588127658633797238=-3035483582895424875, -1600730095770316997=7644364465767360178, -5977653414665598802=6443496393438179244, -4289645355525492039=-7771435519918529374, 4818452273885227082=2145248154554053428, -2465951973599725012=3719500247592761581}, [11182]]
:
:
20536 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.task - Emitting: slidingsum default [16102]
20536 [Thread-25-slidingsum-executor[4 4]] INFO o.a.s.d.executor - TRANSFERING tuple [dest: 5 tuple: source: slidingsum:4, stream: default, id: {-217168626838496871=-2098183848111262192, 6694109964234120893=-2919364166006116209, -490688806946141211=-115866302962242997, 6556174365450512594=3417840899423850921, 4973296703617984132=-2075234239029720284, -8433729321932816312=3438734146522000751, -695350376461229364=5479035516364205453, -1356407922762824927=536244103058094589, -8759945507516006916=-7691606294364721504, -1600730095770316997=5198643672682538868, 1928584781574233931=3233801634595403530, 4818452273885227082=3613752122075445564, 4017617819859316672=-5965638492780984127, -7679610850762262585=8746099621132998817, 3628468721856542322=3234989506915660599, 8041484834072108391=2473403470958482418, -8604773776820158944=4291163489101357389, 5275805877791886609=2224008364377626542, -4383830359476279213=-1613810029185700041, -1641897178290464600=-7937957462653577021, 6730277299577429107=-5906850224979170611, -8115189405407159227=-6807612857900762546, -492827708169915833=-3985992390535713144, 7588127658633797238=7922452426043726560, -3771417496478433111=-1133769369307004904, -5977653414665598802=2226449212304201933, -4289645355525492039=8279576780572003648, 2457695339746807997=-943386263403830945, 7278045793299574088=-3628544844039353718, -2465951973599725012=5019064850597613642}, [16102]]
最後に、3個合計Tupleが溜まったところで平均値を計算して出力しています。
20540 [Thread-23-tumblingavg-executor[5 5]] INFO o.a.s.d.task - Emitting: tumblingavg default [11112]
この後PrinterBoltにデータ送付しているのですが、PrinterBoltは非常にシンプルなので、ここで触れるのは割愛したいと思います。
さて、上記のような感じでスライディングウィンドウも使えました。設定した閾値に伴い指定された動作をしてくれるので、簡単な仕組みであれば、Stormだけで動作させられそうです。
具体的な関数の整備はこれからのようですが、SlidingWindowSumBoltやTumblingWindowAvgBoltを見ればわかるように、「BaseWindowedBoltを実装してexecuteで計算させる」というシンプルでオーソドックスなAPIになっています。EsperやWSO2 CEP(Siddhi)といったCEPプロダクトの関数を実装したことがある人なら、難なく実装できると思います。