前回のブログ記事「Mac OS X環境で、Apache Stormの基本動作を試してみる」のつづきです。
Apache Stormは、multi-language対応なので、"Scala版 storm-starter"を試してみます。
⬛︎ Scala版 storm-starterを構築する
- Githubから、Scala版 storm-starterリポジトリ情報をダウンロードして、ビルドします。
もともとは、paulfryzel/scala-storm-starterのリポジトリ
をベースにして、最新版のStormで動作するように一部修正しております。
$ git clone https://github.com/ttsubo/scala-storm-starter.git
$ cd scala-storm-starter/
$ git checkout sample
$ brew install sbt
$ sbt compile package assembly
- "Nimbus"を起動します
$ storm nimbus
Running: java -server -Ddaemon.name=nimbus -Dstorm.options= -Dstorm.home=/usr/local/apache-storm-0.10.0 -Dstorm.log.dir=/usr/local/apache-storm-0.10.0/logs -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /usr/local/apache-storm-0.10.0/lib/asm-4.0.jar:/usr/local/apache-storm-0.10.0/lib/clojure-1.6.0.jar:/usr/local/apache-storm-0.10.0/lib/disruptor-2.10.4.jar:/usr/local/apache-storm-0.10.0/lib/hadoop-auth-2.4.0.jar:/usr/local/apache-storm-0.10.0/lib/kryo-2.21.jar:/usr/local/apache-storm-0.10.0/lib/log4j-api-2.1.jar:/usr/local/apache-storm-0.10.0/lib/log4j-core-2.1.jar:/usr/local/apache-storm-0.10.0/lib/log4j-over-slf4j-1.6.6.jar:/usr/local/apache-storm-0.10.0/lib/log4j-slf4j-impl-2.1.jar:/usr/local/apache-storm-0.10.0/lib/minlog-1.2.jar:/usr/local/apache-storm-0.10.0/lib/reflectasm-1.07-shaded.jar:/usr/local/apache-storm-0.10.0/lib/servlet-api-2.5.jar:/usr/local/apache-storm-0.10.0/lib/slf4j-api-1.7.7.jar:/usr/local/apache-storm-0.10.0/lib/storm-core-0.10.0.jar:/usr/local/storm/conf -Xmx1024m -Dlogfile.name=nimbus.log -Dlog4j.configurationFile=/usr/local/apache-storm-0.10.0/log4j2/cluster.xml backtype.storm.daemon.nimbus
- "Supervisor"を起動します
$ storm supervisor
Running: java -server -Ddaemon.name=supervisor -Dstorm.options= -Dstorm.home=/usr/local/apache-storm-0.10.0 -Dstorm.log.dir=/usr/local/apache-storm-0.10.0/logs -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /usr/local/apache-storm-0.10.0/lib/asm-4.0.jar:/usr/local/apache-storm-0.10.0/lib/clojure-1.6.0.jar:/usr/local/apache-storm-0.10.0/lib/disruptor-2.10.4.jar:/usr/local/apache-storm-0.10.0/lib/hadoop-auth-2.4.0.jar:/usr/local/apache-storm-0.10.0/lib/kryo-2.21.jar:/usr/local/apache-storm-0.10.0/lib/log4j-api-2.1.jar:/usr/local/apache-storm-0.10.0/lib/log4j-core-2.1.jar:/usr/local/apache-storm-0.10.0/lib/log4j-over-slf4j-1.6.6.jar:/usr/local/apache-storm-0.10.0/lib/log4j-slf4j-impl-2.1.jar:/usr/local/apache-storm-0.10.0/lib/minlog-1.2.jar:/usr/local/apache-storm-0.10.0/lib/reflectasm-1.07-shaded.jar:/usr/local/apache-storm-0.10.0/lib/servlet-api-2.5.jar:/usr/local/apache-storm-0.10.0/lib/slf4j-api-1.7.7.jar:/usr/local/apache-storm-0.10.0/lib/storm-core-0.10.0.jar:/usr/local/storm/conf -Xmx256m -Dlogfile.name=supervisor.log -Dlog4j.configurationFile=/usr/local/apache-storm-0.10.0/log4j2/cluster.xml backtype.storm.daemon.supervisor
- "UI"を起動します
$ storm ui
Running: java -server -Ddaemon.name=ui -Dstorm.options= -Dstorm.home=/usr/local/apache-storm-0.10.0 -Dstorm.log.dir=/usr/local/apache-storm-0.10.0/logs -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /usr/local/apache-storm-0.10.0/lib/asm-4.0.jar:/usr/local/apache-storm-0.10.0/lib/clojure-1.6.0.jar:/usr/local/apache-storm-0.10.0/lib/disruptor-2.10.4.jar:/usr/local/apache-storm-0.10.0/lib/hadoop-auth-2.4.0.jar:/usr/local/apache-storm-0.10.0/lib/kryo-2.21.jar:/usr/local/apache-storm-0.10.0/lib/log4j-api-2.1.jar:/usr/local/apache-storm-0.10.0/lib/log4j-core-2.1.jar:/usr/local/apache-storm-0.10.0/lib/log4j-over-slf4j-1.6.6.jar:/usr/local/apache-storm-0.10.0/lib/log4j-slf4j-impl-2.1.jar:/usr/local/apache-storm-0.10.0/lib/minlog-1.2.jar:/usr/local/apache-storm-0.10.0/lib/reflectasm-1.07-shaded.jar:/usr/local/apache-storm-0.10.0/lib/servlet-api-2.5.jar:/usr/local/apache-storm-0.10.0/lib/slf4j-api-1.7.7.jar:/usr/local/apache-storm-0.10.0/lib/storm-core-0.10.0.jar:/usr/local/apache-storm-0.10.0:/usr/local/storm/conf -Xmx768m -Dlogfile.name=ui.log -Dlog4j.configurationFile=/usr/local/apache-storm-0.10.0/log4j2/cluster.xml backtype.storm.ui.core
- "ExclamationTopology"をローカルモードで動作させます。
$ storm jar target/scala-2.11/scala-storm-starter-assembly-0.0.2-SNAPSHOT.jar storm.starter.topology.ExclamationTopology
...(snip)
6159 [Thread-36-word] INFO b.s.d.task - Emitting: word default [bertels]
6162 [Thread-36-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 4 TUPLE: source: word:10, stream: default, id: {}, [bertels]
6163 [Thread-38-word] INFO b.s.d.task - Emitting: word default [jackson]
6168 [Thread-38-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 3 TUPLE: source: word:14, stream: default, id: {}, [jackson]
6171 [Thread-40-exclaim1] INFO b.s.d.executor - Processing received message FOR 4 TUPLE: source: word:10, stream: default, id: {}, [bertels]
6172 [Thread-18-exclaim1] INFO b.s.d.executor - Processing received message FOR 3 TUPLE: source: word:14, stream: default, id: {}, [jackson]
6174 [Thread-24-word] INFO b.s.d.task - Emitting: word default [jackson]
6174 [Thread-10-word] INFO b.s.d.task - Emitting: word default [bertels]
6174 [Thread-24-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 2 TUPLE: source: word:9, stream: default, id: {}, [jackson]
6174 [Thread-10-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 4 TUPLE: source: word:8, stream: default, id: {}, [bertels]
6174 [Thread-42-word] INFO b.s.d.task - Emitting: word default [jackson]
6174 [Thread-42-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 3 TUPLE: source: word:15, stream: default, id: {}, [jackson]
6174 [Thread-14-exclaim1] INFO b.s.d.executor - Processing received message FOR 2 TUPLE: source: word:9, stream: default, id: {}, [jackson]
6179 [Thread-26-word] INFO b.s.d.task - Emitting: word default [golda]
6179 [Thread-26-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 2 TUPLE: source: word:11, stream: default, id: {}, [golda]
6190 [Thread-12-word] INFO b.s.d.task - Emitting: word default [golda]
6190 [Thread-12-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 3 TUPLE: source: word:12, stream: default, id: {}, [golda]
6200 [Thread-18-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [jackson!!!]
6200 [Thread-14-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [jackson!!!]
6200 [Thread-40-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [bertels!!!]
6201 [Thread-40-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 6 TUPLE: source: exclaim1:4, stream: default, id: {}, [bertels!!!]
6201 [Thread-14-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 5 TUPLE: source: exclaim1:2, stream: default, id: {}, [jackson!!!]
6201 [Thread-18-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 6 TUPLE: source: exclaim1:3, stream: default, id: {}, [jackson!!!]
6201 [Thread-14-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 2 TIME: TUPLE: source: word:9, stream: default, id: {}, [jackson]
6201 [Thread-40-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 4 TIME: TUPLE: source: word:10, stream: default, id: {}, [bertels]
6201 [Thread-14-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:9, stream: default, id: {}, [jackson] TASK: 2 DELTA:
6201 [Thread-40-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:10, stream: default, id: {}, [bertels] TASK: 4 DELTA:
6201 [Thread-18-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 3 TIME: TUPLE: source: word:14, stream: default, id: {}, [jackson]
6201 [Thread-34-exclaim2] INFO b.s.d.executor - Processing received message FOR 5 TUPLE: source: exclaim1:2, stream: default, id: {}, [jackson!!!]
6201 [Thread-40-exclaim1] INFO b.s.d.executor - Processing received message FOR 4 TUPLE: source: word:8, stream: default, id: {}, [bertels]
6201 [Thread-22-exclaim2] INFO b.s.d.executor - Processing received message FOR 6 TUPLE: source: exclaim1:4, stream: default, id: {}, [bertels!!!]
6201 [Thread-40-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [bertels!!!]
6201 [Thread-34-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [jackson!!!!!!]
6201 [Thread-14-exclaim1] INFO b.s.d.executor - Processing received message FOR 2 TUPLE: source: word:11, stream: default, id: {}, [golda]
6201 [Thread-18-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:14, stream: default, id: {}, [jackson] TASK: 3 DELTA:
6202 [Thread-34-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 5 TIME: TUPLE: source: exclaim1:2, stream: default, id: {}, [jackson!!!]
6202 [Thread-14-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [golda!!!]
6202 [Thread-18-exclaim1] INFO b.s.d.executor - Processing received message FOR 3 TUPLE: source: word:15, stream: default, id: {}, [jackson]
6202 [Thread-34-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:2, stream: default, id: {}, [jackson!!!] TASK: 5 DELTA:
6202 [Thread-18-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [jackson!!!]
6202 [Thread-14-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 6 TUPLE: source: exclaim1:2, stream: default, id: {}, [golda!!!]
6202 [Thread-22-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [bertels!!!!!!]
6202 [Thread-40-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 5 TUPLE: source: exclaim1:4, stream: default, id: {}, [bertels!!!]
6202 [Thread-22-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 6 TIME: TUPLE: source: exclaim1:4, stream: default, id: {}, [bertels!!!]
6202 [Thread-18-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 5 TUPLE: source: exclaim1:3, stream: default, id: {}, [jackson!!!]
6202 [Thread-22-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:4, stream: default, id: {}, [bertels!!!] TASK: 6 DELTA:
6202 [Thread-18-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 3 TIME: TUPLE: source: word:15, stream: default, id: {}, [jackson]
6202 [Thread-22-exclaim2] INFO b.s.d.executor - Processing received message FOR 6 TUPLE: source: exclaim1:3, stream: default, id: {}, [jackson!!!]
6202 [Thread-18-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:15, stream: default, id: {}, [jackson] TASK: 3 DELTA:
6202 [Thread-22-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [jackson!!!!!!]
6202 [Thread-18-exclaim1] INFO b.s.d.executor - Processing received message FOR 3 TUPLE: source: word:12, stream: default, id: {}, [golda]
6202 [Thread-14-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 2 TIME: TUPLE: source: word:11, stream: default, id: {}, [golda]
6203 [Thread-14-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:11, stream: default, id: {}, [golda] TASK: 2 DELTA:
6203 [Thread-18-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [golda!!!]
6202 [Thread-40-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 4 TIME: TUPLE: source: word:8, stream: default, id: {}, [bertels]
6203 [Thread-34-exclaim2] INFO b.s.d.executor - Processing received message FOR 5 TUPLE: source: exclaim1:4, stream: default, id: {}, [bertels!!!]
6203 [Thread-18-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 6 TUPLE: source: exclaim1:3, stream: default, id: {}, [golda!!!]
6203 [Thread-40-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:8, stream: default, id: {}, [bertels] TASK: 4 DELTA:
6203 [Thread-34-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [bertels!!!!!!]
6203 [Thread-18-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 3 TIME: TUPLE: source: word:12, stream: default, id: {}, [golda]
6203 [Thread-34-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 5 TIME: TUPLE: source: exclaim1:4, stream: default, id: {}, [bertels!!!]
6204 [Thread-18-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:12, stream: default, id: {}, [golda] TASK: 3 DELTA:
6204 [Thread-34-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:4, stream: default, id: {}, [bertels!!!] TASK: 5 DELTA:
6204 [Thread-22-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 6 TIME: TUPLE: source: exclaim1:3, stream: default, id: {}, [jackson!!!]
6204 [Thread-34-exclaim2] INFO b.s.d.executor - Processing received message FOR 5 TUPLE: source: exclaim1:3, stream: default, id: {}, [jackson!!!]
6204 [Thread-22-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:3, stream: default, id: {}, [jackson!!!] TASK: 6 DELTA:
6204 [Thread-34-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [jackson!!!!!!]
6204 [Thread-22-exclaim2] INFO b.s.d.executor - Processing received message FOR 6 TUPLE: source: exclaim1:2, stream: default, id: {}, [golda!!!]
6204 [Thread-34-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 5 TIME: TUPLE: source: exclaim1:3, stream: default, id: {}, [jackson!!!]
6204 [Thread-22-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [golda!!!!!!]
6205 [Thread-34-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:3, stream: default, id: {}, [jackson!!!] TASK: 5 DELTA:
6205 [Thread-22-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 6 TIME: TUPLE: source: exclaim1:2, stream: default, id: {}, [golda!!!]
6205 [Thread-22-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:2, stream: default, id: {}, [golda!!!] TASK: 6 DELTA:
6205 [Thread-22-exclaim2] INFO b.s.d.executor - Processing received message FOR 6 TUPLE: source: exclaim1:3, stream: default, id: {}, [golda!!!]
6205 [Thread-22-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [golda!!!!!!]
6205 [Thread-22-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 6 TIME: TUPLE: source: exclaim1:3, stream: default, id: {}, [golda!!!]
6205 [Thread-22-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:3, stream: default, id: {}, [golda!!!] TASK: 6 DELTA:
6209 [Thread-30-word] INFO b.s.d.task - Emitting: word default [nathan]
6209 [Thread-16-word] INFO b.s.d.task - Emitting: word default [bertels]
6210 [Thread-16-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 3 TUPLE: source: word:7, stream: default, id: {}, [bertels]
6210 [Thread-30-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 2 TUPLE: source: word:16, stream: default, id: {}, [nathan]
6210 [Thread-18-exclaim1] INFO b.s.d.executor - Processing received message FOR 3 TUPLE: source: word:7, stream: default, id: {}, [bertels]
6210 [Thread-14-exclaim1] INFO b.s.d.executor - Processing received message FOR 2 TUPLE: source: word:16, stream: default, id: {}, [nathan]
6211 [Thread-18-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [bertels!!!]
6211 [Thread-14-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [nathan!!!]
6212 [Thread-14-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 5 TUPLE: source: exclaim1:2, stream: default, id: {}, [nathan!!!]
6212 [Thread-14-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 2 TIME: TUPLE: source: word:16, stream: default, id: {}, [nathan]
6213 [Thread-14-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:16, stream: default, id: {}, [nathan] TASK: 2 DELTA:
6212 [Thread-18-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 5 TUPLE: source: exclaim1:3, stream: default, id: {}, [bertels!!!]
6214 [Thread-18-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 3 TIME: TUPLE: source: word:7, stream: default, id: {}, [bertels]
6215 [Thread-18-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:7, stream: default, id: {}, [bertels] TASK: 3 DELTA:
6215 [Thread-34-exclaim2] INFO b.s.d.executor - Processing received message FOR 5 TUPLE: source: exclaim1:2, stream: default, id: {}, [nathan!!!]
6216 [Thread-34-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [nathan!!!!!!]
6218 [Thread-34-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 5 TIME: TUPLE: source: exclaim1:2, stream: default, id: {}, [nathan!!!]
6219 [Thread-34-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:2, stream: default, id: {}, [nathan!!!] TASK: 5 DELTA:
6219 [Thread-34-exclaim2] INFO b.s.d.executor - Processing received message FOR 5 TUPLE: source: exclaim1:3, stream: default, id: {}, [bertels!!!]
6219 [Thread-34-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [bertels!!!!!!]
6219 [Thread-34-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 5 TIME: TUPLE: source: exclaim1:3, stream: default, id: {}, [bertels!!!]
6219 [Thread-34-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:3, stream: default, id: {}, [bertels!!!] TASK: 5 DELTA:
6221 [Thread-32-word] INFO b.s.d.task - Emitting: word default [golda]
6222 [Thread-32-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 2 TUPLE: source: word:13, stream: default, id: {}, [golda]
6223 [Thread-14-exclaim1] INFO b.s.d.executor - Processing received message FOR 2 TUPLE: source: word:13, stream: default, id: {}, [golda]
6223 [Thread-14-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [golda!!!]
6223 [Thread-14-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 6 TUPLE: source: exclaim1:2, stream: default, id: {}, [golda!!!]
6223 [Thread-14-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 2 TIME: TUPLE: source: word:13, stream: default, id: {}, [golda]
6223 [Thread-14-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:13, stream: default, id: {}, [golda] TASK: 2 DELTA:
6224 [Thread-22-exclaim2] INFO b.s.d.executor - Processing received message FOR 6 TUPLE: source: exclaim1:2, stream: default, id: {}, [golda!!!]
6224 [Thread-22-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [golda!!!!!!]
6224 [Thread-22-exclaim2] INFO b.s.d.executor - BOLT ack TASK: 6 TIME: TUPLE: source: exclaim1:2, stream: default, id: {}, [golda!!!]
6224 [Thread-22-exclaim2] INFO b.s.d.executor - Execute done TUPLE source: exclaim1:2, stream: default, id: {}, [golda!!!] TASK: 6 DELTA:
6266 [Thread-36-word] INFO b.s.d.task - Emitting: word default [bertels]
6266 [Thread-36-word] INFO b.s.d.executor - TRANSFERING tuple TASK: 3 TUPLE: source: word:10, stream: default, id: {}, [bertels]
6266 [Thread-18-exclaim1] INFO b.s.d.executor - Processing received message FOR 3 TUPLE: source: word:10, stream: default, id: {}, [bertels]
6266 [Thread-18-exclaim1] INFO b.s.d.task - Emitting: exclaim1 default [bertels!!!]
6266 [Thread-18-exclaim1] INFO b.s.d.executor - TRANSFERING tuple TASK: 6 TUPLE: source: exclaim1:3, stream: default, id: {}, [bertels!!!]
6266 [Thread-18-exclaim1] INFO b.s.d.executor - BOLT ack TASK: 3 TIME: TUPLE: source: word:10, stream: default, id: {}, [bertels]
6267 [Thread-22-exclaim2] INFO b.s.d.executor - Processing received message FOR 6 TUPLE: source: exclaim1:3, stream: default, id: {}, [bertels!!!]
6267 [Thread-18-exclaim1] INFO b.s.d.executor - Execute done TUPLE source: word:10, stream: default, id: {}, [bertels] TASK: 3 DELTA: 1
6267 [Thread-22-exclaim2] INFO b.s.d.task - Emitting: exclaim2 default [bertels!!!!!!]
以上より、前回と同様に、word(TestWordSpout) → exclaim1(ExclamationBolt) → exclaim2(ExclamationBolt)と動作するストリーム処理において、"bertels” → ”bertels!!!” → ”bertels!!!!!!"のように動作している様子が確認できました。
⬛︎ Storm動作におけるストリーム処理って、どうなっているのか？
(1) Topology定義をさぐってみる
まず、TopologyBuilderクラスからインスタンスを生成するところからはじまります。
そして、Topology構成要素となる"Spout"と"Bolt"を定義することにより、ストリーム処理、すなわち、"Tuple"の振る舞いを定義することになります。
ここでは、word(TestWordSpout) → exclaim1(ExclamationBolt) → exclaim2(ExclamationBolt)と動作するストリーム処理を定義しております。
ExclamationTopology.scala
package storm.starter.topology
import backtype.storm.{ Config, LocalCluster, StormSubmitter }
import backtype.storm.testing.TestWordSpout
import backtype.storm.topology.TopologyBuilder
import backtype.storm.utils.Utils
object ExclamationTopology {
def main(args: Array[String]) {
import storm.starter.bolt.ExclamationBolt
val builder: TopologyBuilder = new TopologyBuilder()
builder.setSpout("word", new TestWordSpout(), 10)
builder.setBolt("exclaim1", new ExclamationBolt(), 3).shuffleGrouping("word")
builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1")
val config = new Config()
config.setDebug(true)
if (args != null && args.length > 0) {
config.setNumWorkers(3)
StormSubmitter.submitTopology(args(0), config, builder.createTopology())
} else {
val cluster: LocalCluster = new LocalCluster()
cluster.submitTopology("ExclamationTopology", config, builder.createTopology())
Utils.sleep(5000)
cluster.killTopology("ExclamationTopology")
cluster.shutdown()
}
}
}
ちなみに、Topologyモードについては、引数を指定しないと、ローカルモードとして動作して、引数を指定すると、リモートモードで動作する実装になっております。
StormSubmitter (Storm 0.10.0-SNAPSHOT API)
(2) Spoutをさぐってみる
先ほどのTopologyにて定義されたSpout"TestWordSpout"は、Githubリポジトリ"apache/storm"のjavaコードにて実装されております。
TestWordSpout.java
package org.apache.storm.testing;
import org.apache.storm.Config;
import org.apache.storm.topology.OutputFieldsDeclarer;
import java.util.Map;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
import java.util.HashMap;
import java.util.Random;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class TestWordSpout extends BaseRichSpout {
public static Logger LOG = LoggerFactory.getLogger(TestWordSpout.class);
boolean _isDistributed;
SpoutOutputCollector _collector;
public TestWordSpout() {
this(true);
}
public TestWordSpout(boolean isDistributed) {
_isDistributed = isDistributed;
}
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
_collector = collector;
}
public void close() {
}
public void nextTuple() {
Utils.sleep(100);
final String[] words = new String[] {"nathan", "mike", "jackson", "golda", "bertels"};
final Random rand = new Random();
final String word = words[rand.nextInt(words.length)];
_collector.emit(new Values(word));
}
public void ack(Object msgId) {
}
public void fail(Object msgId) {
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
@Override
public Map<String, Object> getComponentConfiguration() {
if(!_isDistributed) {
Map<String, Object> ret = new HashMap<String, Object>();
ret.put(Config.TOPOLOGY_MAX_TASK_PARALLELISM, 1);
return ret;
} else {
return null;
}
}
}
nextTupleメソッドは、定期的に呼び出されることにより、Tupleが生成されます。
(3) Boltをさぐってみる
Spout: word → Bolt: exclaim1 → Bolt: exclaim2というTuple処理において、Bolt動作は、ExclamationBoltクラスで定義されております。
ExclamationBolt
package storm.starter.bolt
import backtype.storm.task.{ OutputCollector, TopologyContext }
import backtype.storm.topology.base.BaseRichBolt
import backtype.storm.topology.OutputFieldsDeclarer
import backtype.storm.tuple.{ Fields, Tuple, Values }
import java.util.{ Map => JMap }
class ExclamationBolt extends BaseRichBolt {
var collector: OutputCollector = _
override def prepare(config: JMap[_, _], context: TopologyContext, collector: OutputCollector) {
this.collector = collector
}
override def execute(tuple: Tuple) {
this.collector.emit(tuple, new Values(Exclaimer.exclaim(tuple.getString(0))))
this.collector.ack(tuple)
}
override def declareOutputFields(declarer: OutputFieldsDeclarer) {
declarer.declare(new Fields("word"))
}
}
なお、Bolt:exclaim1では、Tupleを受け取るごとに、executeメソッドが起動されます。
次のBolt:exclaim2に引き渡すにあたって、"!!!"の文字列を追加します。
そして、Tuple処理が完了したことを示す、ackメソッドが起動されます。
Exclaimer.scala
package storm.starter.bolt
object Exclaimer {
def exclaim(s: String): String = {
s + "!!!"
}
}
実際の"!!!"の文字列の追加は、exclaimメソッドで実施されます。
なお、Bolt: exclaim2でも、ExclamationBoltクラスで定義されておりますので、動作は同じになります。
⬛︎ 終わりに
今回、Apache Stormでのmulti-language対応としてをScala版Boltサンプルの動作を確認しました。
Scala版Boltサンプルでも、Apache Stormの基本動作を確認することができました。