Try Storm-Starter

Storm

Storm が気になります！ってわけで、オライリーの Getting Started with Storm を斜め読みして storm-starter を試してみました。

Getting Started With Storm

What is storm-starter?

その名前の通り、Storm をお試しできるプロジェクトです。

• Storm-Starter

必要な環境は Maven もしくは Leiningen のようです。僕は Leiningen の方が慣れてるので、 Lieiningen を選択しました。 （leiningen についてはこちら)

Java と Clojure のコードが用意してありますが、もちろん Clojure 一択です！

Try storm-starter with Leiningen and Clojure

Install

storm-starter は leiningen プロジェクトになっているので、ダウンロードしてきて lein コマンドから依存関係を解決するだけで、動きます。楽チンですね！

$ git clone git@github.com:nathanmarz/storm-starter.git
$ cd storm-starter
$ lein deps
$ lein compile
$ lein run -m storm.starter.clj.word-count

実際に実行してみると、ローカルモードでも並列実行されていることがわかります。

Word Count Program

実行すると文章の単語を数えるプログラムが動きます。大まかに言うと次のような流れになっています。

1. Spout から文が流れる
2. Bolt で単語に分割する
3. Bolt で単語毎に数える

実行されるコードは下記です。

• storm-starter/src/clj/storm/starter/clj/word_count.clj

Clojure DSL によって、Topology, Spout, Bolt が簡潔に記述されているのがわかりますね。 Clojure すごい。

• Clojure DSL

短いコードなので少し見ていきましょう。

main

まずは main 関数です。さっきは引数なしで実行したので、run-local! 関数が実行されています。

（引数をつけて実行すると、それがトポロジー名になってクラスターに送られて、分散モードで実行されますが、分散モードの話はそのうち。）

(defn -main
  ([]
   (run-local!))
  ([name]
   (submit-topology! name)))

main 関数を追うと、トポロジーを作って、10000 ミリ秒実行して止まるようになっています。 その間に Spout から流れた文の単語を数えます。

mk-topology

トポロジーの定義です。

(defn mk-topology []

  (topology
   {"1" (spout-spec sentence-spout)
    "2" (spout-spec (sentence-spout-parameterized
                     ["the cat jumped over the door"
                      "greetings from a faraway land"])
                     :p 2)}
   {"3" (bolt-spec {"1" :shuffle "2" :shuffle}
                   split-sentence
                   :p 5)
    "4" (bolt-spec {"3" ["word"]}
                   word-count
                   :p 6)}))

• “1”と"2" は Spout
• "3"と"4"は Bolt
• “2” が 2 並列、"3" が 5並列、"4" が 6並列
• "1","2" から "3" に渡して、単語ごとに "4" に渡す
• “4” は渡された単語ごとにインクリメントする

たぶんこんな感じです:p

実際に使うとなると、 "1","2" の Spout が Storm の外からデータ入れないといけないのと、"4"" から結果を集めてあげる必要がありそうです。

Multilang protocol

Storm は JVM 系の言語を使うこともできますが、JVMではない言語も使うことができます。 Storm-Starter には先ほどの Word-Count プログラムの split-sentence Bolt の Ruby版と Python版が用意してあります。 こうち Python 版を試してみました。

shell-bolt-spec

Topology を定義する際に、split-sentence Bolt を Python で書かれた Bolt を使うように定義します。

(defn mk-topology []

  (topology
   {"1" (spout-spec sentence-spout)
    "2" (spout-spec (sentence-spout-parameterized
                     ["the cat jumped over the door"
                      "greetings from a faraway land"])
                     :p 2)}
   {"3" (shell-bolt-spec {"1" :shuffle "2" :shuffle}
           　　　　　　　　　"python"
   　　　　　　　　　　　　　　"splitsentence.py"
   　　　　　　　　　　　　　　["word"]
   　　　　　　　　　　　　　　:p 5)
    "4" (bolt-spec {"3" ["word"]}
                   word-count
                   :p 6)}))

実行は先ほどと同じです。

$ lein compile
$ lein run -m storm.starter.clj.word-count

プロセスを監視していると、python のプロセスが 5つ立ち上がることがわかると思います。 Non-JVM な言語と Storm の通信は標準入出力でやりとりされるようです。

まとめ

動くとこまでは結構あっさり行きました。あとは分散モードの環境を作って、ローカルでテストしたトポロジーをデプロイして試していくと楽しそうです。 Storm に乗っけてしまえば、スケールしていきそうなので、アルゴリズムをどうトポロジーに落とし込んで行くかがポイントになりそうです。

あと、Clojure いいですね！Clojure!

Appendix

• Storm Project
• Storm Clojure DSL
• リアルタイム分散処理Stormの耐障害性は？