DEIM Forum 2014 D3-6

Silk による分散データ処理のストリーミング
斉藤 太郎†† 東京大学大学院 情報生命科学専攻
E-mail: †leo@xerial.org
あらまし Silk は並列・分散計算とそのパイプライン処理 (ストリーミング) の融合を目指したフレームワークである。

Silk ではプログラム中のデータ（変数）と分散演算 (map, reduce, join など) の依存関係をプログラムの AST から解析し、クラスタの空き状況に応じて実行プランの最適化やデータの分割を行う。また、プログラムの部分実行により、

GNU Make のように障害からの回復やプログラムの変更後のパイプラインの再計算が可能であり、変数名を指定した
中間データの検索（ワークフロークエリ）を実現できるのが特徴である。本研究では Silk のプログラミングモデルとシステムの設計・実装について紹介し、その応用について議論する。
キーワード

分散データベース、クラウドコンピューティング

1. は じ め に

イエンスの分野における例を題材に説明していく。ゲノムデー
タの処理では、データベース化される前の大量のテキストデー

Silk は並列分散計算とそのパイプライン処理（ストリーミン

タを扱う必要がある。例えば、次世代シーケンサー (Illumina,グ）の融合を目指したフレームワークであり、Scala のライブ

PacBio 社が開発しているもの）と呼ばれる配列解読器の進化

ラリとして実装されている。以下に Silk でのプログラミング例

により A, C, G, T からなる DNA 配列（ゲノム）を低コストを示す。

で大量に読むことができるようになり、病気の原因などを探る

val in = loadFile("input.sam")val size = in.lines.map(_.split("\t")).filter(_(2) == "chr1").size

材料として広く活用されている。シーケンサーから出力される情報は扱いやすいテキスト形式（あるいはそれを圧縮したもの）
が用いられている。ゲノムサイエンスでは各種データ解析のためのプログラムも非常に多く開発されている（興味がある読者

このコードは input.sam というテキスト形式のファイルを開は Bioinformatics 誌、Nature Methods 誌などを参照すると良き (loadFile)、各行を読んで (lines) タブで区切り (split)、い）が、これらのプログラムもテキストベースのフォーマット

2 列目の要素が chr1 となるものを取り出して (filter) その数(FASTA, FASTQ, GFF, BED, SAM, WIG, VCF など) を入をカウントする (size) 計算を表している。ゲノム情報処理で出力に採用している。これは、汎用的な書式でデータのスキーはこのような簡単なデータ処理を日常的に行うことが多い。

マを定義し、研究者間のデータのやり取りを円滑にする文化が小さなテキストファイルでは問題にはならないが、これがコミュニティにあるためである。これにより、データ解析に関100GB を超えるとしたらどうであろう。秀でたプログラマはわる人材のバックグラウンドの多様さ（プログラミングスキル、

マルチコアで計算するためのスレッド処理を加えるであろう

言語の好みなど）に広く対応できるようになっている。

し、分散計算に詳しい技術者なら Hadoop の MapReduce プロ

一方、既存のデータベースシステム側で扱えるデータ形式

グラムに置き換えるかもしれない。データベース技術者ならは非常に狭い。RDBMS、XML DB, NoSQL など種々のシス

RDBMS 用にテキストを整形し、SQL 文で問い合わせを記述

テムがあるが、現状扱えるデータはリレーショナルモデルかドするであろう。

キュメント型（XML、JSON、raw text など) にほぼ絞られてしかし、実行方式は違えど、行いたい計算は既に上記の小さおり、ゲノムサイエンスのデータを扱うためにはフォーマット

なプログラムに記述されている。Silk でのチャレンジは、計算の変換が必須となっている。

環境（マルチコア、クラスタ環境、メモリ容量、共有ファイル

2. 1 巨大なデータの処理システムの有無など）に合わせて、コードはそのままに実行方

データを DBMS で扱える形式にする際、まずデータの大き式のみを切り替えることにある。Silk は分散計算を利用するプ

さが障害となる。ヒトゲノム解析では一人分のゲノムデータを

ログラマ・データサイエンティストの負担を軽減することを念

50x で読む場合（ヒトゲノムのサイズは約 3GB で、エラー補

頭にデザインされたシステムで、実装とその開発はオープン

正のため通常この 50 倍程度の配列データを読む必要がある）、

ソースで GitHub 上 (https://github.com/xerial/silk) で行わ

FASTQ（ゲノム配列＋そのクオリティの値）のテキストデータれている。

で約 500GB にもなる（gzip 圧縮するとこの 1/5 程度のサイズ

2. ゲノムサイエンスにおけるデータ処理
Silk のようなフレームワークが必要になった背景を、ゲノムサになる）。その後、ヒトゲノムにアラインメントする作業で 1.5
倍の SAM データ（テキストで 750GB) が出力される。ETL（extract - trasform - load) でデータベースにデータを取り込—1—

んで解析を行なうにも、まず分散処理が必須なのである。

2. 2 多様なデータ解析
ゲノムサイエンスに置ける解析は他種多様で、SQL の枠に収と誤った結果が生成されることになるため、モジュール化を支援する仕組みも必要である。

2. 5 並列・分散計算習得の難しさ

まらない計算も多く必要となる。上手くデータを RDBMS に取並列・分散計算のプログラムを書くためには、スレッド、プ

り込めたとしても、研究に必要な計算（例えばゲノム配列へのロセスの知識、ロックによる同期、データのメモリ間共有、プアラインメント (参照）、ゲノムアセンブリなど (参照)）は、通ロセス間共有、ソケットなどネットワーク周りの知識などコン常大規模なメモリを必要とし DBMS 側の機能だけでは十分にピューターサイエンスの学部・大学院レベルの多様な知識が要

実装ができない。また、UDF として DBMS 側に新しい関数を求される。分散・並列計算の実行方式でも様々な実装がある。

追加することも考えられるが、実装の複雑さに比べてわざわざ

C/C++なら OpenMP や MPI、Java では Hadoop 関連のエコ

DBMS に組み込む利点が少ない。そのため、DBMS を使わずシステムのプロジェクト群があるが、専門家以外には運用すら済ますよう独自のデータフォーマットの開発が進んでいる。例

難しくなっている。MPI では通信周りの最適化は施されている

えば SAM フォーマット（ゲノムのアラインメント情報のためが、プログラムの書きやすさの面では、例えば一般のオブジェのデファクトスタンダード）のデータを扱う samtools はデー

クトを通信用のデータ構造に使うだけでもオブジェクトのパッ

タのバイナリフォーマット（BAM) への変換、ソーティング、

ケージングが必要となるなど、とたんに記述が難しくなって

インデキシング、区間クエリの実装を独自に行い、各種言語へしまう。IBM が開発している X10 [3] は比較的新しいフレームの移植 (picard ライブラリ, bamtools など) も盛んに行なわれワークでオブジェクトの通信は大分書きやすくはなっているが、ている。このようなデータ処理は本来 DBMS が得意とするは障害対応、パイプライン処理の実行という点は十分にサポートずの機能であるが、多様な解析とデータ処理がうまく融合できされていない。高度な知識を学んでも手軽に使えるフレームていないために DBMS を活用できていない事例である。

ワークが見当たらないのが現状となっている。

2. 3 データのパイプライン処理

分散計算のニーズがゲノムサイエンスのように必ずしもコン

ヒトの疾患関連変異を見つけるための解析では、通常１サンピューターサイエンスを専門に学んだ人以外にも広がっている

プルでは不十分で、数百〜数万規模のサンプルデータを処理し、現在では、コンピューターサイエンスと同等の大学教育を分散

その中での共通変異や特異的な変異を見いだし、病気の原因を計算を行う全ての人に習得してもらうのは現実的ではない。む

探る解析が必要になる。このような計算は１パスで処理しきる

しろこれらの知識を学んだ人が、フレームワークの開発に力ことは難しく、クラスタの障害（ノードの障害、ファイルシス

を注ぎ、分散計算が手軽になるように尽力すべきだと考える。

テムの不具合、プログラムのバグ、メモリ不足）などが起った

データサイエンスを行う人の関心はやはり大きな「謎」を解明

場合でも、再実行できるようにデータ処理をパイプライン化ししたり「発見」する側の「サイエンス」にあり、手段としてのておく必要がある。

分散計算（これも「コンピューターサイエンス」として大きなまた病気の種類により追加で必要な解析も変わってくる。遺伝性の疾患の場合、疾患関連変異を持っていても必ずしも発症する訳ではなく（浸透率）、弧発性疾患の場合は家系を調べても病気の原因がわからない。ゲノムだけではなく、エピゲノム(ゲノム上に乗るデータ全般）のデータ（メチル化、ヒストン
修飾など）の状態を次世代シーケンサーを使って調べることもテーマであることに変わりないが）に費やす労力をできるだけ軽減することに価値があるだろう。

3. Silk フレームワーク
前説までで述べたように、データ処理のプログラムには以下に挙げる課題があるあり、そのような場合にはさらに解析パイプラインが複雑にな•

巨大なデータの処理る。解析結果を見て結果が思わしくなければ、パイプラインの•多様なデータ解析プログラムの融合

変更、プログラムのパラメータのチューニングなどの処理を施•

解析のモジュール化・パイプライン化

すことも多く、パイプライン処理といっても定型処理ではなく、•障害・拡張への対応

研究のニーズに応じて拡張されていくものになる。

•

習得のしやすさ2. 4 解析プログラムのモジュール化

Silk ではプログラムモジュール化の手間を軽減するため、

ゲノムサイエンスに限らずデータサイエンスではプログラミ

light-weigt なモジュールとして、プログラミング言語中の関数

ングとデータ処理が密結合している。しかし、実際のデータ解をそのままモジュールとして使えるように設計されている。モ析のパイプラインを構築する現場ではコードをモジュール化しジュールの依存関係は、Silk では関数の依存関係として表され疎結合にするアプローチが取られてきた。例えば、UNIX のコる。プログラムのモジュールも関数も入力を受け取って出力をマンド群を組み合わせて Makefile のようなパイプラインを書く

返すというパラダイム上では同じものである。Scala の簡潔なのがその一例である。タスク間の依存関係はモジュール間の依

シンタックスを利用することにより習得も容易になる。Silk の存関係で表されるが、データサイエンスでは一度しか使わない

データ処理の関数を用いるだけなら Scala の複雑な機能まで全

コードも多く書かれ、プログラム毎にモジュール化するコストて学習する必要はない。

の方が大きくなってしまう。さらに、モジュール間の接続（例

Scala は Java 仮想マシン（JVM) 上で動く言語で、強力なえば受け渡しするファイル名）やデータの保存形式を間違える

コレクションライブラリ（list, set, map などの基本データ構

—2—

関数
in.map[B](f:A=>B)
in.flatMap[B](f:A=>SilkSeq[B])in.filter(f:A=>Boolean)in.sort(implicit ord:Ordering[A])in.reduce(f:(A,A)=>A)
in.sizeleft.join[B,K](right:SilkSeq[B],lk:A=>K, rk:B=>K)
in.groupBy[K](k:A=>K)
in.mapWith[B,R](resource:Silk[R])(f:A=>B)
in.splitin.concat
in.shuffle[K](f:A=>K)c"(unix command)"c"(unix command)".linesc"(unix command)".file(name)

演算オブジェクト
MapOp(in:SilkSeq[A], f:A=>B)
FlatMapOp(in:SilkSeq[A], f:A=>SilkSeq[B])FilterOp(in:SilkSeq[A], f:A=>Boolean)
SortOp(in:SilkSeq[A], ord:Ordering[A])ReduceOp(in:SilkSeq[A], f:(A,A)=>A)
SizeOp(in:SilkSeq[A])JoinOp(left:SilkSeq[A], right:SilkSeq[B],lk:A=>K, rk:B=>K)
GroupByOp(in:SilkSeq[A], k:A=>K)
MapWithOp(in:SilkSeq[A], resource:Silk[R], f:A=>B)
SplitOp(in:SilkSeq[A])ConcatOp(in:SilkSeq[SilkSeq[A]])ShuffleOp(in:SilkSeq[A], f:A=>K)
CmdOp(cmd:String, inputs:Array[Silk[ ]])CmdOutputLines(cmd:StringContext, inputs:Array[Silk[ ]])CmdOutputFile(cmd:StringContext, file:String,
inputs:Array[Silk[ ]])結果の型
SilkSeq[B]SilkSeq[B]SilkSeq[A]SilkSeq[A]SilkSingle[A]SilkSingle[Long]

SilkSeq[(A, B)]
SilkSeq[(K, SilkSeq[A])]SilkSeq[B]SilkSeq[SilkSeq[A]]SilkSeq[A]SilkSeq[(K, SilkSeq[A])]SilkSingle[Any]SilkSeq[String]
SilkSingle[File]

表 1 Silk の演算オブジェクト（抜粋）。入力データ in に対して関数を呼び出した結果生成される演算オブジェクトとその型を示している。全ての演算オブジェクトにはその結果が代入される変数名や定義されているクラス名、行番号などの情報 (fcontext) が含まれているが表記上は省略している

造）を備えているため、データ処理分野への応用が進んでいる。

中に map, filter, join などの分散演算が現れた箇所周辺の AST

Scala にバージョン 2.9 から標準で搭載されている並列コレクを探索し、各々の演算結果がどの変数に代入されているかを見ション [15] は、マルチコア環境での並列データ処理をスレッドつけることができる。例えば、

プログラミングの知識無しに利用できる。分散環境への対応でval a = input.map(f)は、Spark の分散データセット (RDD) [19] が提案されておりコミュニティによる開発も盛んである。Twitter 社も Hadoopのジョブを簡易に記述するための Scala API である Scaldingを公開している。

Scala のシンタックスは簡潔になりやすいため、同等のデー
タ処理コードを Java で書くのに比べてコード量が大幅に短くなる傾向がある。これはプログラミングの専門家以外にとってラーニングコストを下げる上で利点となる。また、関数型言語の特徴である高階関数 (関数に関数を渡す) の記述が自然にサポートされているため、MapReduce のように関数をリモートに配備し分散で実行させるスタイルによく合致している。

Silk が目標にするのは、Scala で記述されたパイプラインの拡張をしながら、障害の発生時には、既に計算が済んだ部分の結果は再利用しつつパイプラインの残りの処理を実行する仕組みである。これにはプログラムの途中結果を保存する仕組みと、プログラムの部分実行をサポートする必要がある。次節からこれらの機能をどう実現していくかについて解説する。

という Silk の演算は、マクロでコンパイル時に以下のような演算オブジェクトに置き換えられる。

val a = MapOp(in:input, out:"a", op:f)
ここで MapOp は map を表す演算オブジェクト、in は入力デー
タ、out は結果の出力先の変数名の文字列、op は in の各要素に対して適用される関数である。
実際のシステムでは、変数名だけでなく、変数が定義されている関数名、クラス名、ソースコード中の行番号などより詳細な情報が AST から取得されており、これをプログラム中のマーカーとして用いている。演算オブジェクトの実行結果を、
このマーカー名を用いてキャッシュやディスクに保存することで、プログラム中の変数名を用いてワークフローを流れるデー
タの検索や参照が可能になる。

3. 2 ネストした演算
前説の変数 a の結果から、10 より大きい要素を取り出す式を追加すると、以下のようになる。

3. 1 コードへのマーキングプログラム中の変数名を使って途中結果を保存できれば、変数名を使ってプログラムの途中結果に対するクエリを実行する
ことができ、さらに、変数名を指定して再実行したいプログラval b = a.filter(_ > 10)
ここで変数 a、b に代入される演算オブジェクトは、

ム中の箇所を選択できるようになる。ここでいかにプログラム

val a = MapOp(in:input, out:"a", op:f)val b = FilterOp(in:a, out:"b", op:{_ > 10})中の変数名を取得するかが問題になる。変数名の情報は通常コであり、実際にこの変数 b の中身を展開すると以下のようにネンパイル時には失われてしまい、実行時のプログラムは知るこ

ストした演算オブジェクトなっている。

とができない。
計算結果がどの変数に代入されたかの情報を得るために、コン
パイル時に得られる情報を活用するアプローチに至った。2013
年に Scala に導入されたマクロ [1] の機能を用いると、関数呼び出しの記述をコンパイル時に書き換えることが可能になる。
その際、関数呼び出し周辺の構文木（AST, Abstract Syntax

Tree) を取得することができる。この機能を用いると、コード

FilterOp(
in:MapOp(in:input, out:"a", op:f),
out:"b",op:{_ > 10})
Silk の全ての演算は、このネストを許した演算オブジェクトで表される。表 1 に Silk で定義される演算オブジェクトの一覧を示す。これは関係代数 (relational algebra) における操作—3—

命令 (σ(selection), π(projection) など) に該当する。演算オブ

Silk には演算オブジェクトを評価する数種類の weaver が提

ジェクトは Silk[A] という型を持つ。これは A というオブジェ

供されている。メモリ中で計算を済ませる InMemoryWeaver, ク

クト型を生成する演算オブジェクトであることを表し、単一のラスタ環境で分散してデータ処理をする ClusterWeaver, GNU

オブジェクトを生成する演算には SilkSingle[A] 型、リスト

Make の用に結果を逐次ファイルに書き出す MakeWeaver があ型を生成する演算には SilkSeq[A] 型が付けられている。Join

る。この他にも後から独自の weaver を追加することも可能な演算など入力が２つ以上の場合もあり一般には DAG(directed

ように設計している。

acyclic graph) の形状を取る。将来的に SQL と同等の演算オ

3. 6 静的な最適化

ブジェクトを Silk に導入することも検討している。

Silk[A] の DAG スケジュールはパターンマッチによりグラ

3. 3 UNIX コマンドの利用

フを組み替えることで、ルールベースの最適化を施すことがで

Silk では UNIX コマンドのパイプラインを作成することもできる。この静的な最適化の例としては、関数の合成ルール

きる。以下の例は、カレントディレクトリ内の*.txt ファイルを

map(f).map(g) => map(f andThen g)ls コマンドで列挙し、各々に対して wc(word count) コマンドを実行し、単語数（のリスト）を計算する例である。

val inputFiles = c"ls -1 *.txt".lines
val wordCounts =for(file <- inputFiles)
yield c"wc $file".lines.head

や、map と filtering の合成により生成される中間オブジェク
トの量を減らすルール

map(f).filter(c) => mapWithFilter(f, c)などがある。他にも Silk 内で使われるストレージの種類により

c で始まる文字列 c (UNIX command)"は UNIX コマンドのpushing-down selection、columnar ストレージへのアクセス、

演算オブジェクトを表し、この文字列内には変数や演算など

オブジェクトの部分的な projection などの適用が考えられる。

の任意の式を$expr の形式で埋め込める。for 文は Scala では

for-comprehension と呼ばれ、inputFiles の各要素 file に

4. クラスタでの実行

対し、yield 以下の式を実行した結果を連結したものを返す。

クラスタで分散演算を実行するための ClusterWeaver の概

for-comprehension は map 演算の syntax sugar であり、上記の要を図 1 に示す。以降の節では、ClusterWeaver の各コンポー

for 文のコードは

ネントについて紹介し、どのように Silk が分散演算を実行してinputFiles.map(file => c"wc $file".lines.head)

いるかについて解説する。

4. 1 ClassBox: リモートでのコード実行と書くのと等しい。この例の場合は、各ファイル毎の wc コマ
ンドの実行結果の先頭行のリストになる。

UNIX コマンドの実行結果が標準出力に返される場合は.lines で各行のデータを取得できる。ファイルに結果を書き込む場合は、.file(name) でファイル名を明示的に返すことで、コマンド実行結果のファイルを受け取ることができる。

3. 4 演算オブジェクトの拡張
Silk では関数を再利用できるため、上記の基本演算に加えて独自の実装を追加することもできるようになっている。例えば、

Silk での hash-join の実装は、表 1 の演算の組み合わせ (map,shuﬄe, reduce など) と、データがメモリに十分収まるサイズになったときにインメモリで join を実行する記述になっている。大規模データのソーティングでは、データから数%程度のサンプルを取得しヒストグラムを作成し、ノード間の偏りが無いようにデータを分散してソートしているが、このヒストグラ
ムの作成も Silk のパイプラインで記述されている。

3. 5 Weaving: 演算オブジェクトの評価
Silk[A] の演算オブジェクトは weave 関数を呼び出すと、
スケジュールの評価が始まる。これは、素材 (silk) を編んで

(weave）織物 (product) を生成するのに例えている。Silk の設計では、Silk[A] の構築と、weaving の過程を明確に切り分けている。そうすることで、パイプラインの記述は変えずに、実際に使う計算機環境に合わせて実行方式だけを切り替えられるよ
うにできる。これはパイプラインのテスト用と本番環境での実Silk は Scala 上で動くコードであり、Java プログラムと同様 JVM 上で動く。リモートに立ち上げた SilkClient(JVM で動作）でローカル JVM と同じようにコードを実行するためには、まず classpath に含まれる全ての class ファイルと jar ファ
イルをリモートノードにも配備する必要がある。Hadoop や

Spark では必要な jar をコード中に指定する必要があるが、Silk
では、ローカル環境中の classpath と jar のエントリを列挙しClassBox と呼ばれるオブジェクトを生成する。ClassBox にはclasspath エントリと jar ファイルのパスとその hash 値の情報が含まれている。ClassBox をリモートに転送（実際にはリモー
トの SilkClient が jar ファイルを pull する）することでローカルと同じ環境をリモートノードで再現できる。

UNIX コマンドを各ノードで同じように実行するには、VMの利用や、docker などの軽量 UNIX コンテナなどを利用する
より汎用性の高い方法が考えられるが、ここでは簡単のため全ノードで同じ UNIX コマンドが使えることを想定している。

4. 2 クロージャのクリーンアップ
MapOp の演算オブジェクトではリモートで関数 f:A=>B（Aの型のデータを引数に受け取り B の型の結果を返す関数）を実行する必要がある。そのために関数 f の内部で参照されている
自由変数 (free variable. 関数の引数で定義されていない変数）
すなわち、関数の外で定義されている外部変数 (outer variable)のデータを提供する必要がある。以下の例では、変数 N が mapに渡される関数内で定義されていないので外部変数となる。

行の切り替えにも役立つ。

—4—

Local machineWeaving Silk materializes objects

Local ClassBox	
 classpaths & local jar files
User program
builds workflows

•
•
•
•
•

Silk[A]	

Silk[A]	

read file, toSilk
map, reduce, join,groupBy
UNIX commands
etc.

SilkSingle[A]	
 SilkSeq[A]	

weave
weave

Static optimization

DAG Schedule	

••

Cluster••••Dispatches tasks to clientsManages master resource table
Authorizes resource allocation
Automatic recovery byleader election in ZK

Register ClassBox
Submit schedule

A	

Seq[A]	

single object

sequence of objects

ZooKeeper	
ensemble mode
(at least 3 ZK instances)

Silk Master	••dispatch

••

Silk Client	

Silk Client	

Task Scheduler	

Task Scheduler	

Task Executor	

Task Executor	

Resource Monitor	

Resource Monitor	

Data Server	

Data Server	

Node Table	
Slice Table	

Task Status	
Leader election
Collects locations of slices
and ClassBox jars
Resource Table
Watches active nodes(CPU, memory)	
Watches available resources
ClassBox Table	
•
•
•
•
•
•
•
•

Submits tasks
Run-time optimization
Resource allocation
Monitoring resource usage
Launches Web UIManages assigned task status
Object serialization/deserialization
Serves slice data

図 1 Silk の概要図. Silk では手元のローカルマシンで Silk を使った演算を記述する。Silk[A]のデータはそのままプログラムの DAG を構成している。ルールベースの最適化によりDAG スケジュールを生成し、クラスタ中で動作している SilkMaster にスケジュールを送り Silk の結果を評価するタスクを実行する。Silk クラスタの維持に重要な情報はレプリカ構成で３ノード以上で動作する ZooKeeper に保存されており、SilkMaster が落ちた
場合には他の SilkClient が SilkMaster に昇格する。SilkClient は個々のタスクの実行管理、スケジューリング、リソース確保の役割を担う。メモリやローカルディスク上のデー
タを配信するための DataServer も各 SilkClient で立ち上がっており HTTP ベースで各ノードとデータのやり取りができるようになっている。

val N = 100
val result = in.map(_ * N)にリモートに送るバイナリサイズが巨大になってしまう問題が// N は外部変数ある。

この N の情報と関数の情報同時に送らないとリモートで正し

実際に関数中で使われる最小限の外部変数の情報を得るため

い計算が実行できない。この外部変数の環境と関数本体を合わに、Silk では関数 f のバイトコードをスキャンしており、そのせたものをクロージャーという。

操作のために ASM (Java の bytecode を操作するためのライ

例えば Hadoop では DistributedCache(バージョン 2.x 系列ブラリ) を用いている。関数の動作を JVM のスタックへの操では JobContext) に外部変数の情報を登録しておき、全ノード

作も含めてシミュレートすることで、関数 f からどのような関からアクセスできるようにしているが、クロージャー内の外部数が呼び出されるかを追跡でき（データフロー解析）、コンテ

変数の出現を把握できればこのプロセスは自動化できる。

キスト内で定義されていない変数にアクセスした際にそれを外Scala ではクロージャーは１つのクラスとして定義されてい部変数への参照として記録する。この操作で各関数（クラス名）

るため、ClassBox が転送済みであれば、外部変数の情報を補と、その内部で使われる外部変数の対応表ができる。この対応

うことでクロージャーをリモートでも実行できる。しかしその表の作成は、関数１つにつき１回実行するだけで良いため重い

クロージャー内で参照されているクラスのうち１つのフィール計算ではない。この情報を元に、使われないフィールドを null

ドしかアクセスされない場合、クロージャーのシリアライズ時値でクリアすることで、Java の標準のシリアライザーを使って—5—クロージャーをシリアライズした場合でもバイナリサイズを必

グのアプローチを取っている。マルチスケジュール最適化など

要最小限に納められるようになった。

を行う際には、単一のマスターがリソースの共有に基づいた

また、Silk では大きな外部変数を必要があって使う場合は、外部変数として参照するではなく、mapWithResource 関数

in.mapWithResource(resource:R)(f:A=>B)

最適化を行う方が都合が良いが、それと同時にマスターの負荷が重くなりすぎインタラクティブな解析に弱くなる欠点がある。Silk ではより細粒度のタスクが多いことを想定し分散スケ
ジューラーのモデルを採用している。

の利用を推奨している。mapWithResource では、外部変数の代
わりに f の内部で使用できるリソース R の情報を渡すことで、リモートに送るデータサイズを小さくしており、リソースのあるノードに近い場所でなるべく計算を行うなどの工夫ができるようになっている。

4. 3 クラスタのセットアップ
Silk でのクラスタのセットアップはできるだけ簡単になる
ように実装されている。ユーザーが最低限設定に必要なのは、$HOME/.silk/hosts ファイルにクラスタとして使うノードのホ
スト名を一行毎に列挙するだけで良い。silk cluster start
コマンドを実行すると、SilkClient を指定されたホストに ssh
ログインして立ち上げてくれる。hosts ファイルがない場合は単一ノードで動く standalone モードで Silk が動作する。

4. 4 ZooKeeper
Silk ではノードの死活管理や分散データのインデキシング、タスクの実行状態の記録などの用途に ZooKeeper [7] を用いている。ZooKeeper はクラスター計算機のためのコーディネー
ションシステムで、ノード間の同期を取るためのプリミティブな操作（分散ロックなど）が提供されており、比較的小さなデー
タの共有データストレージとしての役割も果たす。Silk ではク
ラスタの起動時にノードのリストから３台を選び、ZooKeeper
を ensemble 構成で自動的に起動するようにしている。これによりユーザーは ZooKeeper をセットするアップ手間を省けるようになるとともに、ノードの障害時に ZooKeeper 内のデー
タ喪失も防ぐこともできる。
各 ノ ー ド で 起 動 さ れ た SilkClient は 、ZooKeeper のephemeral mode（ノードへの接続が維持されている間だけデータが存在する）を用いて自身の情報（ホスト名、アドレス、
ポート番号、プロセス ID など）を記録しており、クラスタ中
でアクティブなノードのリストが ZooKeeper 上で管理されるようになっている。

ZooKeeper は leader election の 機 能 も 果 た し て お り、
SilkClient の中から１台を SilkMaster に選出するために使われている。SilkMaster が何らかの理由でクラッシュした
場合は、SilkClient の中から新しい SilkMaster が選出され、ZooKeeper などに保存された情報から自動的に復帰するように設計されている。

4. 5 SilkMaster
SilkMaster はタスクの SilkClient への配分 (dispatch)、リ
ソーステーブルのマスターの保持などの役割を担う。Silk[A] を

ClusterWeaver 上で実行する際にはまず SilkMaster に DAG
スケジュールを送信することになる。SilkMaster が DAG ス
ケジュールを受け取ると、アクティブな SilkClient の中から１台を選んでスケジューリングを委譲する分散スケジューリン

4. 6 SilkClient
SilkClient は各ノードでデーモンして立ち上げられており、タスクのスケジューリング、実行、ノードの CPU 使用率、メ
モリ使用量の管理を行っている。

4. 7 Data Server
Silk では SilkSeq[A] 型の大きなデータを Slice という単位に分割し、各ノードにばらまくことで分散計算を実現している。その Slice のデータを保持するのが DataServer である。DataServer は各 SilkClient 毎に配備され、Slice がどの

DataServer に保存されているかの情報は ZooKeeper に記録されている。インメモリ計算の場合は DataServer にはオブジェ
クトデータがそのまま保存されており、ローカルノードからのリクエスト時にはそのままオブジェクトの参照を渡すが、リモートからデータがリクエストされた場合には、オブジェクトをシリアライズして転送する。

4. 8 Resource Monitor
Silk では、クラスタ中のあらゆるジョブが単一スケジュー
ラーで管理されることを想定しておらず、他のジョブ実行エン
ジンと共存することを前提にしている。これは、共用クラス
タでは、他のジョブがノードの CPU やメモリを埋め尽くしている場合もあり、この状態を把握するために、単一スケジュー
ラーで全てを動かす（例えば Mesos [5] や、YARN [17] のアプローチ）に移行するよりは、各ノードでリソースモニターを動かし、実際に使用されている CPU の load average や空きメモ
リ量を定期的に ZooKeeper に記録し、その情報を活用するよ
うにしている。

4. 9 Task Scheduler
タスクのスケジューリングは各 SilkClient が独立に行っている。これは Google の Omega スケジューラー [16] に倣った設計で、リソースの使用状況のテーブルを、クラスタ中の全ノードが共有できる前提になっている。
リソースの使い方の判断は、Spark で使用されている delayscheduling [18] と同様に、データの locality を考慮して、計算に必要なデータを持っているノードを優先的に使用する。もし何回かの試行（閾値とタイムアウトを設定）でノードが利用可
能にならなければ、他の利用可能なノードにデータを転送して計算を実行される割り当てを採用する。
リソースの使い方を決めた後は SilkMaster に問い合わせをし承認を得る。実際のリソースの空き状況と大きくかけ離れていなければ許可を与える設計になっており、楽観的な判断を行っている。これは CPU を１つ使うと宣言したタスクで合っても、CPU を必ずしも 100%使用するわけではないことへの対応である。厳密なリソース割り当てを行うと、12 コアのノード
では高々12 個のタスクしか行えないが、現実にはその 1.5 倍程—6—

度のタスクを与えてもうまく処理できることが多い。タスクのである。この利点のみを活かし、タスク間 (inter-task)、タス

割り当て過ぎへの対処には、各ノードから集計された実際のリ

ク内 (intra-task) 並列・分散実行を行えるような枠組みを Silk

ソース使用量の情報をもとにつじつまを合わせ、タスクの配分

では採用している。

の許可を調整するようにしている。

5. 今後の展望と課題
Silk の開発は現在も進んでおり、今後、現場での活用、大規模な性能評価を実施することで性能・機能の向上につながることを期待している。ここでは Silk の活用法について議論する。

以下に Makefile と Silk の機能の対比を示す。
機能

Makefile

Silk

タスク間の依存関係

ファイルの引数 Silk 型の変数の参照

タスクの実行

コマンドの実行 Silk 型の変数の評価

タスクの細分化ファイルの分割 SilkSeq の分割

表 2 Makefile と Silk の機能の対比

5. 1 データ解析のサイクルの効率化「ワークフローのプログラミング」-「分散計算の実行」- 「結5. 4 Iterative computing への対応

果の確認」-「コードの修正・拡張」-「再計算」、このサイクル

PageRank, K-means の計算などはループによる処理を含み、の効率を向上するためには全ての処理をパイプライン化する必

iterative computing と呼ばれる。現在のところ Silk ではループ要がある。まずデータ処理のための数行のコードを Silk で記

毎に weave 処理を繰り返すことで iterative computing を実現

述してクラスタ環境にコードをデプロイし、実行をスタートさしているが、itertive computing の代数的枠組みへの導入も鬼せる。結果を確認するためのコードを書くのに一番手軽な方法

塚らにより提案されており [14]、Silk への導入も検討している。

は、ソースコード中にコードを追加することであろう。Silk で

5. 5 LArray: Oﬀ-Heap ストレージ

は変数名を参照して変数の内容を確認するなど任意の操作を追

分散インメモリ計算ではメモリストレージが JVM の Out-加できるため、データが出力されたファイル名を指定するなど

OfMemoryException の発生や、GC 処理の負荷のために応答の手間がない。俗に printf デバッグと呼ばれる方式であるが、が停止してしまうという問題がある。この問題を解決するにJava 界隈でも slf4j など文字列を出力する API しかない logger

は、JVM が管理するヒーブ外の領域 (oﬀ-heap) でデータを確が広く使われている。breakpoint などを使用するデバッガと違保し、データが不要になったら即座にメモリから解放できる仕いプログラムの動作を手動で止める必要がないという利便さも組みが必要となる。Java の標準にはこのようなメモリ管理をある。

実現できる仕組みがないため、Oﬀ-heap メモリを扱うライブ

5. 2 ワークフロークエリ

ラリ LArray (https://github.com/xerial/larray) を開発しリ

実際にパイプラインで解析中の途中結果に対してクエリを投リースしている。

げることで、どのようなデータが生成されているかをモニタリ

5. 6 圧縮オブジェクトストレージ

ングすることも可能になる。データの様子を確認することで、

Silk の分散演算では大量のオブジェクトが生成できるように

次にどのような解析をすれば良いかのアイデアも生まれるようなる。このオブジェクトデータをスライス毎に効率よく圧縮し、になり、次の解析のためのコードがパイプラインに追加されるディスクなどに保存して永続化するストレージの実装も進めてようになる。Silk では、変数名を使った中間結果へのマーキンいる。オブジェクトを列分解し、gzip, snappy などで圧縮する

グを行うことで、計算済みの結果へのアクセス方法を提供してことでデータサイズを小さくすることができるが、列分解する

おり、ワークフローに対して変数名を使ったクエリを記述する

コストや、オブジェクトへの復元の高速化が課題となっている。

方向性も期待できる。

5. 3 Makefile の代替としての利用

6. 関 連 研 究

このようなデータ解析のワークフローを実現するのに同等の巨大なデータの処理には、データをクラスタ内に分散配置

ものとして Makefile が挙げられる。Makefile では UNIX コマ

して並列処理する MapReduce やその実装である Hadoop がンドのパイプラインの記述ができ、通常ソースコードのコンパ

広く用いられるようになった。一般のデータ処理では単一のイル方法の指示などに使われる。Makefile の成果物はファイル

MapReduce だけでは十分ではなく、データ処理のパイプライ

であり、必要なファイルが存在しなければそのファイルを生成ンを作成するために、Pig [13]、Oozie、Twitter Summingbird

するためのルールを Makefile 中の記述から見つけ出し、生成などのオープンソースプロジェクトが登場している。SQL 言語に必要な UNIX コマンドを実行する仕組みである。

を Hadoop 上で実現するために Hive [6], 分散ストレージへのMakefile はデータサイエンスに置けるパイプラインの記述にアクセスをプラグインにし SQL の演算部分のみを切り出したも応用することもできるが、Makefile におけるタスクの実行

分散クエリ処理エンジンである Facebook の Presto などのプ

単位は UNIX コマンド毎にタスク間の並列・分散実行 (例えばロジェクトも登場している。

Sun grid engine や GXP などを用いる) は可能であるが、タス

Pig [13] では複数の MapReduce 処理を組み合わせたパイプラ

ク内並列化はできない。また入出力のデータがファイルになる

インを独自のスクリプト言語を用いて記述する。FlumeJava [2]ため、インメモリコンピューティングによる計算の高速化に応では Java 言語で MapReduce のパイプラインを手軽に記述する

用することが難しい。

ためのプリミティブ演算を提供するライブラリであるが、Scala

Makefile を使う利点はタスク間の依存関係を記述できることを用いるほどの簡潔さはなく、Hadoop と密結合された実装で—7—あるため実行エンジンの切り替えが難しい。DryadLINQ [4] はかつ、運用の手間を減らし、多様なマシン構成に対応できるよ

プログラミング言語中で SQL ライクな構文を使えるようにすうにするのはチャレンジングであるが、取り組む価値のある問ることで、計算とデータベースへのアクセスを融合させている。

題である。

Nova [12] は Hadoop のジョブをパイプライン化したものに対し、入力データに更新が起った場合、再計算が必要な部分のみの差分を逐次更新するためのフレームワークである。Microsoftの Niad ではこれを diﬀerential workflow [11] と定義し、同等の問題を扱っている。
分散計算の iterative computing への対応では、Spark [19] が各ノードのメモリ中データをキャッシュしておくことで処理の効率を挙げている。Spark は Silk と似ている部分があるが、プログラム全体の DAG スケジュールを構築するのではなく、mapなどの関数ごとにスケジュールを構築する逐次評価のアプローチが取られている。このため、プログラム全体の最適化、実行方式の切り替えを行うにはプロジェクトの大規模な書き換えが必要になるため、Silk では独自の実装を用いる判断に至った。
低レイテンシでのクエリの実行には Google の Dremel [10]や、Muppet [9]、Cloudera の Impala などがある。Dremel は
カラム指向ストレージを活用し、アクセスするデータ量の削減、データの集約演算の高速化を実現している。
分散環境でのジョブスケジューラーには、データの局所性を考慮するものが提案されている。Dryad の Quincy [8] ではグラフを用いた最適化が行われ、Spark では delay scheduling [18]が用いられている。これらのアプローチは HDFS など shared-

nothing 型の構成でデータを転送するネットワークのコストが重い場合に重要であるが、近年価格が下がってきた Infiniband
では 10Gbps を超える速度でデータを転送できる。このようなネットワークが普及してくると、ボトルネックはデータの局所性ではなく、オブジェクトのシリアライズなどによる CPU に推移すると考えられ、スケジューリングアルゴリズムも再考が必要になるだろう。
クラスタで多数のサービスを同時に動かしている場合、そのリソース管理には Mesos [5] や Apache の YARN [17] などのリソースマネージャーの活用が進んでいる。Mesos では単一のマスターが各ノードに対してリソースを提供 (oﬀer) する方式を取っており、Hadoop や Spark [19] などのフレームワーク側が、それらのリソースをどのように使用するかを決定できる。ただし、これらのリソースマネージャーではクラスタ全体のリソー
ス状況をフレームワーク側が知ることができないため、Google
では Omega [16] スケジューラーを開発し、リソーステーブル
を全ノードが共有するアプローチを採用し、サービス型、タス
ク型とリソースの利用時間が異なるジョブを共存できるようにしている。

7. お わ り にSilk は分散データ処理を書きやすくし、パイプラインの最適化・拡張・部分計算を実現するためにデザインされたシステムである。Silk のターゲットは、データを使ったサイエンス・分
析を行う人すべてであり、ゲノムサイエンスなど応用できる場
面は多岐に渡る。分散計算への敷居や学習コストを下げ、なお文

献[1] E. Burmako and M. Odersky. Scala Macros, a Technical
Report. EPFL, 2012.
[2] C. Chambers, A. Raniwala, F. Perry, S. Adams, R. R.
Henry, R. Bradshaw, and N. Weizenbaum. FlumeJava: easy,
eﬃcient data-parallel pipelines. ACM SIGPLAN Notices,45(6):363–375, 2010.
[3] P. Charles, C. Grothoﬀ, V. Saraswat, C. Donawa, A. Kielstra, K. Ebcioglu, C. von Praun, and V. Sarkar. X10: An
object-oriented approach to non-uniform cluster computing.
SIGPLAN Not., 40(10):519–538, Oct. 2005.
[4] Y. Fetterly, M. Budiu, Ú. Erlingsson, and e. al. DryadLINQ:
A System for General-Purpose Distributed Data-Parallel
Computing Using a High-Level Language. Proc. LSDS-IR,
2009.
[5] B. Hindman, A. Konwinski, M. Zaharia, A. Ghodsi, A. D.
Joseph, R. Katz, S. Shenker, and I. Stoica. Mesos: A platform for fine-grained resource sharing in the data center. In
proc. of NSDI, pages 22–22, 2011.
[6] Apache Hive. http://hive.apache.org.
[7] P. Hunt, M. Konar, F. P. Junqueira, and B. Reed.
ZooKeeper: Wait-free coordination for Internet-scale systems. USENIX ATC, 10, 2010.
[8] M. Isard, V. Prabhakaran, J. Currey, U. Wieder, K. Talwar,
and A. Goldberg. Quincy: fair scheduling for distributed
computing clusters. In proc. of SOSP ’09, Oct. 2009.
[9] W. Lam, L. Liu, S. Prasad, A. Rajaraman, Z. Vacheri, and
A. Doan. Muppet: MapReduce-style processing of fast data.
proc. of VLDB, 5(12):1814–1825, Aug. 2012.
[10] S. Melnik, A. Gubarev, and e. al. Dremel: Interactive Analysis of Web-Scale Datasets. In proc. of VLDB, 2010.
[11] S. R. Mihaylov, Z. G. Ives, and S. Guha. REX: recursive,delta-based data-centric computation. In proc. of VLDB,
July 2012.
[12] C. Olston, G. Chiou, L. Chitnis, F. Liu, Y. Han, M. Larsson, A. Neumann, V. B. N. Rao, V. Sankarasubramanian,
S. Seth, C. Tian, T. ZiCornell, and X. Wang. Nova: continuous Pig/Hadoop workflows. In SIGMOD ’11, June 2011.
[13] C. Olston, B. Reed, U. Srivastava, R. Kumar, andA. Tomkins. Pig latin: a not-so-foreign language for dataprocessing. In proc. of SIGMOD, pages 1099–1110, 2008.
[14] M. Onizuka, H. Kato, S. Hidaka, K. Nakano, and Z. Hu.
Optimization for iterative queries on mapreduce. PVLDB,
7(4), 2013.
[15] A. Prokopec, P. Bagwell, T. Rompf, and M. Odersky. A
Generic Parallel Collection Framework. In Euro-Par. 2011.
[16] M. Schwarzkopf, A. Konwinski, M. Abd-El-Malek, and
J. Wilkes. Omega: flexible, scalable schedulers for large
compute clusters. In proc. of EuroSys, pages 351–364,
Prague, Czech Republic, 2013.
[17] V. K. Vavilapalli, A. C. Murthy, C. Douglas, S. Agarwal,
M. Konar, R. Evans, T. Graves, J. Lowe, H. Shah, S. Seth,
B. Saha, C. Curino, O. O’Malley, S. Radia, B. Reed, and
E. Baldeschwieler. Apache hadoop YARN: Yet another resource negotiator. In proc. of SOCC ’13, 2013.
[18] M. Zaharia, D. Borthakur, J. Sen Sarma, K. Elmeleegy,
S. Shenker, and I. Stoica. Delay scheduling: a simple technique for achieving locality and fairness in cluster scheduling. In EuroSys ’10, Apr. 2010.
[19] M. Zaharia, M. Chowdhury, T. Das, A. Dave, J. Ma, M. McCauley, M. J. Franklin, S. Shenker, and I. Stoica. Resilient distributed datasets: a fault-tolerant abstraction forin-memory cluster computing. In NSDI’12, Apr. 2012.

—8—