読者です 読者をやめる 読者になる 読者になる

夢とガラクタの集積場

落ちこぼれ三流エンジニアである管理人の夢想=『夢』と、潰えた夢=『ガラクタ』の集積場です。

Stormの内部実装を解説する資料確認してます(その1

こんにちは。

最近StormのGitHubにおいて、『Stormの内部構造の解説ページ』が開設されています。
そのため、ソースコードリーディング・・・とは違いますが、
解説ページを読んで、その内容をまとめていこうと思います。

1.Structure of the codebase

======
Stormのcodebaseは3レイヤーから成る。

1レイヤー目:
マルチ言語対応のレイヤー。
StormのNimbusはThrift serviceであるし、TopologyもThriftの構造をとっている。
そのため、Stormは様々な言語からThriftを通して扱うことが可能となっている。

2レイヤー目:
Stormの各コンポーネントに対するJavaインタフェースレイヤー。
Stormのベース部分はClojureで実装されているが、JavaAPIから扱えるよう
インタフェースを確保している。

3レイヤー目:
Stormのメイン部分のClojure実装レイヤー。
Line数においてはJavaClojureは同程度となっている。
だが、Clojureはラインあたりの集約度が高いため、大部分の実装はClojureで記述されていることとなる。
======

御存じのとおり、Stormの大部分のロジックはClojureで記述されており、
Javaプログラマにとっては『容易に扱うことはできるが、内部解析は困難』というプロダクトとなっています。

とまぁ、ただ最近はClojureだけでなくScala、GroovyといったJVM言語を組み合わせて
OSSプロダクトを構築している例も多いため、それほど特殊ではないとは思います。
Clojureを習得しましょう、というだけですね^^;

では、各レイヤの詳細が続きます。

storm.thrift

======
storm.thrift
Stormではこの「ThriftFork」を用いてコードを生成している。
このForkは「Thrift 7」のパッケージをorg.apache.thrift7に変更したものとなっており、動作としては全く同じである。

Forkを生成した理由として、Thriftが過去バージョンとの互換性がなくなっているため。
ことなるバージョンのThriftからもStorm Topologyが操作できるよう確保している形となっている。

Stormでは各SpoutとBoltに「ComponentID」というユーザが定義する識別子を割り振っている。
ComponentIDはSpout/Bolt→BoltのどのStreamを取得するかの識別に用いられる。
StormのTopologyはComponentID→各Spout/Boltのマッピングを保持している。

SpoutとBoltは同様のThrift定義を保持する。(参照
ComponentObjectとComponentCommonという構造体を保持する。

■ComponentObject
ComponentObjectはBoltに対応する実装定義となっている。下記3つのうちいずれかを取る。

  • 1. IBoltを実装したJavaObject(シリアライズ後)
  • 2. 非JVM言語での実装を示すShellComponentオブジェクト

StormはShellComponentを基に非JVMプロセスとの通信を扱うShellBoltを生成している。

  • 3. Storm上でのクラス名とJavaオブジェクトとしての初期化に必要なコンストラクタの引数を示すJavaObject構造

JVM言語でTopologyを定義する際有用となる構造。この構造によって、JVM上のSpout/Boltを非JVM言語から生成/シリアライズ化して保持させることが可能となっている。

■ComponentCommon
ComponentCommonは下記の通りComponentObject以外のBolt/Spoutの定義を保持する。

  1. 対象コンポーネントがどのStreamに出力するか? + ストリームのメタデータ(ダイレクトストリームか、Field declaration)
  2. 対象コンポーネントがどのStreamを取得するか?(StreamGroupingを利用するためのcomponent_id>stream_idのマップ定義)
  3. 対象コンポーネントの並列度(ハッシュ値算出などに用いる)
  4. 対象コンポーネント固有の設定項目

Spout向けの構造体もComponentCommonを保持するため、Boltと同様に『Streamを受信』することは可能である。定義上は。
ただし、StormのJavaAPI上現状Spoutが他Streamを取得する設定は出来ず、
値として設定した場合にもTopologyのSubmit時にエラーとなるようにしている。
理由としては、SpoutはStormのFramework側が使用するStreamの取得定義を保持しているため。

StormはTopology上でAckingFrameworkを利用するためのBoltとStreamを保持しており、
各SpoutはAcker BoltからTupleTreeが処理完了したか、失敗したかを示すAck or Failのメッセージを取得している。
これらのTopologyへのAckerFramework追加のコードはここ

======

つまりはComponentObjectが実際に動作するインスタンスを示して、
ComponentCommonはComponentObjectが他Taskとの通信定義などを保持している・・・
という一般的なモデルを示しているわけですね。

また、ComponentObjectのShellComponent/JavaObjectより、
JVM言語プロセスとの通信や、非JVM言語プロセスからのJVMプロセス定義を重点的にサポートしているのがわかります。

後は、Storm Topologyの内部動作を知る上でおそらく最も重要となる要素のうちの一つ、Acker Frameworkについて。
とりあえず文章から見るとSpoutにしかAcker FrameworkのInputが設定されないかのように書いてありますが、
実際にClojureコード上では全Taskに対してAcker FrameworkのStreamを追加しているようにみえるわけでして・・・はてさて。
尚、Acker Frameworkの他にStorm System Streamなるものも追加を行っています。

とまぁ、この辺りは現在まだページは作成されていませんが、「Acking framework implementation」の章を楽しみにしましょうということで。

次回以降更に続きを確認します。