Hadoop MapReduceのパイプラインジョブ開発 のためのCascadingと

Hadoop MapReduce のパイプラインジョブ開発
のための Cascading と Pig の比較
CAMP システムグループ
遠山 敏章
概要
Hadoop MapReduce は大規模データの分散処理を容易に書くことができるフレーム
ワークである。しかしながら、現実のアプリケーションは複数の MapReduce ジョブ
が連なるパイプラインの実装が必要になる。生の MapReduce によるパイプライン
の実装はコンポーネントの再利用が難しく、コードの保守は煩雑になりがちである。
そこで、本研究レポートでは、Hadoop 上でパイプラインを容易に実装できるソ
フトウェア Cascading と Pig の比較を行う。比較を行うにあたり、それぞれのソフ
トウェアを用いてパイプラインを実装した。そして、「拡張／再利用の容易さ」「テ
ストの容易さ」
「開発サイクルの俊敏さ」の３つの観点から比較した。今回、実装し
たアプリケーションは、検索ログからセッション時間のレンジごとに検索回数の統
計量（平均、分散、最大、最小、中央値）を算出するパイプラインである。
実装を通した比較の結果、Cascading は「拡張／再利用の容易さ」と「テストの
容易さ」に長けており、複雑なパイプラインの実装にも適していた。一方、Pig は
「開発サイクルの俊敏さ」に長けており、データフロー言語である Pig Latin の実行
環境上でインタラクティブにシンプルなパイプラインを実装するのに適していた。
第 1 章 はじめに
Hadoop MapReduce(MR) は大規模データの分散処理を容易に書くことができるフ
レームワークである。コモディティなサーバーをクラスターに追加するだけでスケー
ルアウトでき、安価に大規模なデータ処理が行うことができる。近年では、Mahout
などの機械学習ライブラリの登場により、高度な機械学習も Hadoop 上で容易に行
う事ができるようになり、ログの解析から機械学習まで多様なアプリケーションを
Hadoop で実装できるようになった。アプリケーションの多様化に伴い、Hadoop 上
で実行されるジョブ数は増加している。
複数のジョブで構成される MR のアプリケーションにおいては、パイプラインの
実装が必要になる。パイプライン開発は、次の二つの実装を行う。
1. 個々のジョブの実装 計算処理を MR のプログラミングモデルに落とし込み、複数の MR ジョブを
実装する。そして、MR ジョブを組み合わせて、パイプラインを構成する。複
数のジョブで共通の処理がある場合は処理の共通化を行う。例えば、「グルー
ピングキー」や「集計の時間の粒度」だけが異なり、mapper と reducer の実
装が同じ場合は、GroupingComparator と Partitioner の実装をパラメータ化
する。また、パイプラインの最適化として、複数のジョブを１つにまとめて実
装する。
2. ジョブの依存関係の管理 個々のジョブの入出力とジョブの依存関係を管理し、ジョブフローの調整を
行う。
数個のジョブにおいては、難なくパイプラインを実装できるが、ジョブの数が増
えるにつれてより非常に煩雑な実装になる。また、「ジョブ間の処理の共通化」と
「パイプラインの最適化」を適用した場合、プログラムの可読性が下がり、パイプラ
インの個々のジョブの挙動を理解するのが困難になる。現実のアプリケーションで
は、75 個のジョブで構成されるパイプラインが存在する [2]。10 個以上のジョブで
構成されるパイプラインはジョブフローの管理だけでも煩雑になるため、生の MR
で実装するのは現実的ではない。
そこで、本研究レポートでは MR のパイプライン開発を容易にするソフトウェア
Cascading と Pig の比較を行う。これらのソフトウェアは、Hadoop 上に抽象化層を
提供することで、MR のプログラミングモデルで考える事なく、MR ジョブを実装
できる。パイプラインの一連のジョブを実行した場合、ジョブ間の一時データの管
理は抽象化層で隠蔽され、ユーザーは煩雑なジョブの入出力の管理や依存関係の調
整から解放される。また、カスタム関数のインターフェースを実装することで、再
利用可能な関数を実装できる。パイプラインの最適化においても抽象化層で自動で
行われる。
本研究レポートの構成は次の通りである。第 2 章では、パイプラインの開発を容
易にするソフトウェアについて述べる。第 3 章では、比較を行うために実装した検
索ログの解析アプリケーション [1] について述べる。第 4 章では、アプリケーション
の実装の経験をもとにソフトウェアの比較を行い、第 5 章で結論を述べる。
第 2 章 パイプラインの開発を容易にす
るソフトウェア
本章では、MR のパイプラインの開発を容易にするソフトウェアとして、Cascading,
Pig, Crunch について述べる。
1
Cascading
Cascading は Hadoop 上に抽象化層を実装した Java の API である。データ処理の
ワークフローを定義し実行するための Query API と Query Planner を提供する。
Cascading の処理モデルは pipe と filter をもとにしている。Cascading API はデー
タストリームを split, merge, group、または、join するパイプラインを束ねて、個々
のデータやデータのグループに対して処理を適用する。入出力は Tap によって入
力元の source と 出力先の sink を指定でき、データソースはローカルのファイルシ
ステム、hdfs, Amazon S3 からを選べる。データのレコードは tuple で表現され、
tuple の各データはフィールド名を持つ。Cascading はターゲティング広告、ログ解
析、バイオインフォマティクス、機械学習、ETL アプリケーションに用いられてい
る [3]。
コード 第 2 章.1 にワードカウントの例を示す。
コード 第 2 章.1: Cascading によるワードカウント
1
2
3
// 入 力 元 の source tap の 定 義
Scheme sourceScheme = new TextLine ( new Fields ( " line " ) );
Tap source = new Hfs ( sourceScheme , inputPath );
4
5
6
7
// 出 力 先 の sink tap の 定 義
Scheme sinkScheme = new TextLine ( new Fields ( " word " , " count " ) );
Tap sink = new Hfs ( sinkScheme , outputPath , SinkMode . REPLACE );
8
9
10
// 先 頭 の pipe assembly の 定 義
Pipe assembly = new Pipe ( " wordcount " );
11
12
13
14
15
16
17
// 行 ご と に 文 字 列 を 単 語 に 分 割 す る 。
// 入 力 tuple の line フ ィ ー ル ド を 正 規 表 現 で 単 語 に 分 割 し 、
// word フ ィ ー ル ド を 持 つ tuple と し て 出 力 す る 。
String regex = "(? <!\\ pL )(?=\\ pL )[^ ]*(? <=\\ pL )(?!\\ pL )";
Function function = new RegexGenerator ( new Fields ( " word " ) , regex );
assembly = new Each ( assembly , new Fields ( " line " ) , function );
18
19
20
// word フ ィ ー ル ド で tuple ス ト リ ー ム を グ ル ー ピ ン グ す る 。
assembly = new GroupBy ( assembly , new Fields ( " word " ) );
21
22
23
24
// 全 て の tuple グ ル ー プ に 対 し て 、 単 語 の 出 現 頻 度 を カ ウ ン ト し
// count フ ィ ー ル ド と し て 保 存 す る 。
Aggregator count = new Count ( new Fields ( " count " ) );
25
assembly = new Every ( assembly , count );
26
27
28
29
// ア プ リ ケ ー シ ョ ン プ ロ パ テ ィ を 初 期 化 し 、 Hadoop に 使 用 す る jar を 設 定 す る 。
Properties properties = new Properties ();
FlowConnector . s e t A p p l i c a t i o n J a r C l a s s ( properties , Main . class );
30
31
32
33
// source と sink tap を 使 用 す る assembly か ら 新 し い flow を 作 成 す る 。
FlowConnector flowConnector = new FlowConnector ( properties );
Flow flow = flowConnector . connect ( " word - count " , source , sink , assembly );
34
35
36
// flow を 実 行 し 、 完 了 ま で 待 つ 。
flow . complete ();
2
Pig
Pig は Hadoop 上でデータフローを実行するエンジンである。Pig にはデータフ
ローを表現するデータフロー言語 Pig Latin が用意されており、Pig Latin の join,
sort, filter といった演算子を用いて、処理を記述する。また、独自のデータの読み書
きや処理をするためにユーザー定義関数（UDF）を Java で書くことができる。コー
ド第 2 章.2 にワードカウントの Pig Latin を示す。
Pig には Grunt という組み込みのシェルがあり、ユーザーは Pig Latin をインタ
ラクティブに実行できる。開発とデバッグ用の環境として、ローカルモード (pig -x
local) があり、ローカルのファイルシステムに対して、Pig Latin を実行することが
できる。コード第 2 章.3 にローカルモードでワードカウントを実行した例を示す。
コード 第 2 章.2: Pig Latin によるワードカウント
1
2
-- sentence . txt の 各 行 を フ ィ ー ル ド 名 text の 文 字 配 列 と し て 読 み 込 む
loaded = load ’ sentence . txt ’ as ( text : chararray );
3
4
5
-- 各 行 を f o r e a c h で 読 み 込 み 、 T O K E N I Z E 関 数 で 単 語 に 分 割 し 、 T O K E N I Z E の 戻 り 値 で あ る 「 複
数 の 単 語 」 の 集 合 を flatten で 「 単 語 」 に す る 。
tokenized = foreach loaded generate flatten ( TOKENIZE ( text )) as word ;
6
7
8
-- 単 語 ご と に グ ル ー ピ ン グ す る 。
grouped = group tokenized by word ;
9
10
11
-- COUNT 関 数 で グ ル ー ピ ン グ し た 単 語 を カ ウ ン ト し 、 単 語 を word フ ィ ー ル ド , 単 語 数 を
count と す る 。
counted = foreach grouped generate group as word , COUNT ( tokenized ) as count ;
12
13
14
-- counted を wordcount デ ィ レ ク ト リ に 出 力 す る
store counted into ’ wordcount ’;
コード 第 2 章.3: ローカルモードによる Grunt の実行ログ
1
2
3
4
5
6
bash -3.2 $ pig -x local
12/02/04 19:54:03 WARN conf . Configuration : DEPRECATED : hadoop - site . xml found in the
classpath . Usage of hadoop - site . xml is deprecated . Instead use core - site . xml ,
mapred - site . xml and hdfs - site . xml to override properties of core - default . xml ,
mapred - default . xml and hdfs - default . xml respectively
2012 -02 -04 19:54:03 ,573 [ main ] INFO org . apache . pig . Main - Logging error messages to
: / private / tmp / pig
2012 -02 -04 19:54:03 ,719 [ main ] INFO org . apache . pig . backend . hadoop . executionengine .
HExecutionEngine - Connecting to hadoop file system at : file :/// grunt > loaded =
load ’ sentence . txt ’ as ( text : chararray );
grunt > tokenized = foreach loaded generate flatten ( TOKENIZE ( text )) as word ;
grunt > grouped = group tokenized by word ;
7
8
9
10
11
grunt > counted = foreach grouped generate group as word , COUNT ( tokenized ) as count ;
grunt > store counted into ’ wordcount ’;
2012 -02 -04 19:54:17 ,477 [ main ] INFO org . apache . pig . tools . pigstats . ScriptState - Pig
features used in the script : GROUP_BY
2012 -02 -04 19:54:17 ,477 [ main ] INFO org . apache . pig . backend . hadoop . executionengine .
HExecutionEngine - pig . usenewlogicalplan is set to true . New logical plan will
be used .
2012 -02 -04 19:54:17 ,581 [ main ] INFO org . apache . pig . backend . hadoop . executionengine .
HExecutionEngine - ( Name : counted : Store ( file :/// private / tmp / work / wordcount : org .
apache . pig . builtin . PigStorage ) - scope -16 Operator Key : scope -16)
12
13
... < 省 略 > . . .
14
15
16
17
18
19
20
2012 -02 -04 19:54:20 ,216 [ Thread -16] INFO org . apache . hadoop . mapred . LocalJobRunner reduce > reduce
2012 -02 -04 19:54:20 ,216 [ Thread -16] INFO org . apache . hadoop . mapred . Task - Task ’
attempt_local_0001_r_000000_0 ’ done .
2012 -02 -04 19:54:24 ,882 [ main ] WARN org . apache . pig . tools . pigstats . PigStatsUtil Failed to get RunningJob for job job_local_0001
2012 -02 -04 19:54:24 ,885 [ main ] INFO org . apache . pig . backend . hadoop . executionengine .
mapReduceLayer . MapReduceLauncher - 100% complete
2012 -02 -04 19:54:24 ,885 [ main ] INFO org . apache . pig . tools . pigstats . PigStats Detected Local mode . Stats reported below may be incomplete
2012 -02 -04 19:54:24 ,886 [ main ] INFO org . apache . pig . tools . pigstats . PigStats - Script
Statistics :
21
22
23
HadoopVersion
0.20.2 - cdh3u2
19:54:24
PigVersion
0.8.1 - cdh3u2
GROUP_BY
UserId StartedAt
FinishedAt
toshiaki_toyama 2012 -02 -04 19:54:17
Features
2012 -02 -04
24
25
Success !
26
27
28
29
Job Stats ( time in seconds ):
JobId
Alias
Feature Outputs
job_local_0001 counted , grouped , loaded , tokenized
:/// private / tmp / work / wordcount ,
GROUP_BY , COMBINER
file
30
31
32
Input ( s ):
Successfully read records from : " file :/// private / tmp / work / sentence . txt "
33
34
35
Output ( s ):
Successfully stored records in : " file :/// private / tmp / work / wordcount "
36
37
38
Job DAG :
job_local_0001
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
2012 -02 -04 19:54:24 ,888 [ main ] INFO org . apache . pig . backend . hadoop . executionengine .
mapReduceLayer . MapReduceLauncher - Success !
grunt > quit
bash -3.2 $ head -n 10 ./ wordcount / part -r -00000
a
1
It
1
an
1
as
2
in
1
of
1
on
1
to
1
Pig
3
and
1
3
Crunch
Crunch[4] は MR のパイプラインのコーディング、テスト、実行を容易にする Java
のライブラリである。Crunch の設計は Google の FlumeJava[5] に行われ、シンプル
で少数のプリミティブの演算子と軽量なユーザー定義関数で構成される。プリミティ
ブの演算子とユーザー定義関数を組み合わせることで複雑で、複数のステージのパ
イプラインを構築することができる。Crunch はパイプラインを一連の MR ジョブ
にコンパイルし、ジョブの実行を管理する。Crunch は、執筆時、開発フェーズのた
め本研究レポートでは比較対象のソフトウェアには含めない。Cascading との比較
については、Quora の記事 [6] を参照されたい。
第 3 章 検索ログの解析アプリケー
ション
検索ログの解析アプリケーションは、検索ログからセッション時間のレンジ（ex. 0
∼10 秒、10∼20 秒）ごとに検索回数の統計量（平均、分散、最大、最小、中央値）
を算出する。少なくとも、三つの MR ジョブで構成されるパイプラインである。プ
ログラムは GitHub[1] に公開してある。
1
検索ログ
解析対象の検索ログに Excite の検索クエリーログ*1 を用いた。各行にセッション
ID、タイムスタンプ、検索クエリーの３つのフィールドがタブ区切りで含まれてい
る（コード第 3 章.1）。
コード 第 3 章.1: 検索ログのサンプル
1
2
3
4
5
6
7
8
9
10
11
< sessionId > < timestamp > < searchQuery >
20 BF 184809 garter belts
20 BF 184939 lingerie
20 BF 185051 spiderman
20 BF 185155 tommy hilfiger
20 BF 185513 calgary
20 BF 191233 exhibitionists
184 A 103534 brookings
184 A 104751 breton liberation front
184 A 115238 breton
...
2
↑
｜
｜ ← session1 : sId1
= "20 BF "
｜
sTime1 = 191233 - 184809
｜
= 6424
↓
↑
｜ ← session2 : sId2
= "184 A "
↓
sTime2 = 115238 - 103534
= 11704
解析アプリケーション
解析アプリケーションは検索ログをインプットとし、１．セッション解析、２．統
計値の算出を行う。
１．セッション解析 「セッション ID が同じ一連のレコード*2 」を１セッションと
し、各セッションのセッション時間と検索回数を出力する。コード第 3 章.1 の例で
は、セッション ID(sId) が 20BF である６つの一連のレコードを１セッションとし、
*1
Pig のチュートリアルに同梱されている。
厳密には、セッション ID が同じで、且つ、前後のレコードのタイムスタンプの差が６０秒以内
の一連のレコード
*2
最後と最初のレコードのタイムスタンプの差を計算し、6424 秒のセッション時間
(sTime) を出力する。検索回数はセッションを構成するレコードの件数をカウント
して出力する。
２．統計値の算出 各セッションをセッション時間のレンジごとにグルーピングし、
グループごとに検索回数の統計値を算出する。１つのグループのセッション時間の
間隔は 10 秒とし、「0 秒以上 10 秒未満」のグループ、「10 秒以上 20 秒未満」のグ
ループ、
「20 秒以上 30 秒未満」のグループ．
．
．のようにセッション時間のレンジでグ
ルーピングする。コード第 3 章.1 の例では、session1 は 6 秒のセッション時間なの
で、
「0 秒以上 10 秒未満」のグループに属し、session2 は 11 秒なので「10 秒以上 20
秒未満」のグループに属する。セッション時間のレンジのグルーピング後、検索回
数の統計値の算出を行う。統計値は、平均、分散、中央値、最大値、最小値である。
3
パイプラインの設計
2 で説明した処理を MapReduce のプログラミングモデルに落とし込むと図 3.1 の
ようになる。丸は処理内容、矢印はデータの入出力を表す。このパイプラインは三
つの MR のジョブで構成される。各 MR では以下の処理を行う。
MR1: 丸 1.1 ∼ 1.4 検索ログを入力し、セッション解析を行い、セッション時間と検索回数を出力
する。
図. 3.1: 検索ログの解析アプリケーションのパイプライン
MR2: 丸 2.1 ∼ 2.3 MR1 の出力を入力し、セッション時間のレンジ (sessionRange) ごとに検索回
数の統計値の一部を出力する。出力する統計値は、平均、最大、最小である。
MR3: 丸 3.1 ∼ 3.3 セッション時間のレンジを持つ MR1, MR2 の出力を入力し、セッション時間
のレンジごとに検索回数の分散と中央値を算出し、統計値（平均、分散、中央
値、最大、最小）を出力する。
統計値を計算する際、分散のみを MR3 で算出している理由は、分散*3 を計算するの
には事前に平均を計算しておく必要があるためである。実装の便宜上、中央値の算
出を MR3 に含めたが、MR2 でも算出可能である。
3.1
Cascading でのパイプラインの実装
Cascading では MapReduce の抽象化層として、Function, Filter, Aggregator, Buffer
を用意している。今回はこの抽象化層の中で、Reducer の処理を独自実装できる Buffer
インターフェースを次の二つのクラスで実装した。
AnalyzeSession.java: 丸 1.4 セッションの timestamp のリストをインプットとし、セッションを解析し、検
索回数とセッション時間を出力
CalcStats.java: 丸 3.2 MR2 の sessionRange 毎の統計値の出力と MR1 のセッション毎の検索回数を
インプットとし、検索回数の分散と中央値を計算する。MR2 で算出した統計
値に分散と中央値のフィールドを追加して出力
次に Pipe クラスの実装について述べる。Cascading では Pipe クラスを組み合わせる
事で複雑なパイプラインを実装することができる。実装した Pipe では、パイプライ
ンのデータフローの制御、キーのグルーピングの指定、データの変換の指定を行っ
た。個々の実装の概要は以下の通りである。
GroupedTimeInsertionPipe.java: 丸 2.1 各セッションの sessiontTime から sessionRange を算出してレコードに追加す
るパイプ。
SessionAnalysisPipe.java: 丸 1.2 ∼ 1.4 グルーピングキーとソート順を指定し、AnalyzeSession で処理するように指定
するパイプ
*3
N 個の変数 x からなる平均値 x の分散 σ 2 は σ 2 =
1
N
∑N
k=1 (xk
− x)2 で求められる。
StatsCalculationPipe.java: 丸 2.1 ∼ 3.2 MR1 の出力結果のセッション毎の検索回数と sessionTime をインプットとし、
最終出力の統計値を出力する処理を指定するパイプ。内部で GroupedTimeInsertionPipe を使用。平均、最大、最小の統計値は組み込みの関数で算出した。
Main クラスの SearchSessionStatsMain では入力元と出力先の指定と、実装した
Pipe の組む処理を実装している。
Cascading では、実装したクラスのテストをするためユーティリティクラス CascadingTestCase クラスが用意されている。今回は、Buffer クラスを継承した AnalyzeSession クラスと CalcStats クラスをこの CascadingTestCase クラスを用いてテ
ストした。
3.2
Pig でのパイプラインの実装
Pig では、独自の処理を実装できるユーザー定義関数 (UDF) と PigLaten スクリプ
トの実装を行った。実装を行うにあたり DataFu[7] を参考にした。
UDF の実装においては、EvalFunc を継承し、レコードを跨いで処理を行うため
の Accumulator インターフェースを実装した。実装したクラスは以下の通りである。
ビジネスロジックは Cascading の Buffer のものと同じである。
AnalyzeSession.java: 丸 1.4 セッションの timestamp のリストをインプットとし、セッションを解析し、検
索回数とセッション時間を出力
CalcStats.java: 丸 3.2 MR2 の sessionRange 毎の統計値の出力と MR1 のセッション毎の検索回数を
インプットとし、検索回数の分散と中央値を計算する。MR2 で算出した統計
値に分散と中央値のフィールドを追加して出力
PigLatin スクリプトの実装をコード 第 3 章.2 に示す。PigLatin スクリプトではデー
タフローの定義と実装した UDF の呼び出しを行っている。このビジネスロジックは
Cascading で実装した Pipe クラスと Main クラスのビジネスロジックと同じ内容と
なっている。平均、最大、最小の算出は Pig の組み込み関数を用いた。
コード第 3 章.2 の PigLatin スクリプトのテストは PigTest を用いて行った。
コード 第 3 章.2: 検索ログの解析アプリケーションの PigLatin の実装
1
2
3
register ’ stats - pig -0.0.1 - SNAPSHOT . jar ’;
define analyzeSession pig . AnalyzeSession ();
define calcStats pig . CalcStats ();
4
5
% default g r o u p i n g _t i m e _ g r a n u l a r i t y 10
6
7
8
-- 検 索 ロ グ の 読 み 込 み
records = load ’ excite - small . log ’ as ( sid : chararray , time : long , word : chararray );
9
*3
Cascading の JobPlanner が組んだジョブの実行計画を dot ファイルで出力できる。
10
11
12
13
14
15
16
17
-- セ ッ シ ョ ン ID sid で グ ル ー ピ ン グ
grouped = group records by sid ;
analyzed = foreach grouped {
-- タ イ ム ス タ ン プ の 昇 順 に ソ ー ト
sorted = order records by time ;
-- セ ッ シ ョ ン を 解 析
generate flatten ( analyzeSession ( sorted . time ));
};
18
19
20
-- セ ッ シ ョ ン 時 間 か ら session_range を 算 出 し 、 フ ィ ー ル ド に 追 加
rcds = foreach analyzed generate record_count , ( session_time - ( session_time % (
$ g r o u p i n g _ t i m e _ g r a n u l a r i t y * 1000))) as session_range ;
21
22
23
24
-- 平 均 、 最 小 、 最 大 を 算 出
grouped = group rcds by session_range ;
ret1 = foreach grouped generate group as session_range : long , COUNT ( rcds . record_count
) as count , AVG ( rcds . record_count ) as average , MIN ( rcds . record_count ) as min ,
MAX ( rcds . record_count ) as max ;
25
26
27
28
-- 中 央 値 と 分 散 の 算 出
joined = cogroup ret1 by session_range , rcds by session_range ;
ret2 = foreach joined generate group , flatten ( calcStats ( ret1 , rcds ));
29
30
31
-- 統 計 値 の 書 き 出 し
store ret2 into ’ output / search_session_stats ’;
第 4 章 考察
第 3 章で述べた検索ログの解析アプリケーションの実装の経験をもとに生の MapReduce と Cascading/Pig の比較、そして、Cascading と Pig の比較を行う。
1
生の MapReduce と Cascading/Pig の比較
前述した通り、生の MR の実装によるパイプラインの開発においては、個々のジョ
ブの実装とジョブの依存関係の管理が煩雑になる。Cascading/Pig のソフトウェア
を利用する事でこれらの煩雑な作業から解放される。得られた利点は以下の通りで
ある。
入出力データの自動管理 ジョブ間のデータの入出力の管理はソフトウェア内で管理され、開発者から隠
蔽されている。
組み込み関数の活用 キーのグルーピング、ソートといった MapReduce の基本的な操作から、平均、
最大、最小といった計算まで組み込みの関数が提供されている。これらのライ
ブラリを用いる事で独自に実装する必要がなくなる。
ジョブの実行計画の最適化 Cascading と Pig ではジョブの実行計画の最適化が自動で行なわれる。
一方、ソフトウェアの利用によって、新たな複雑さが生じた。それは、タプルのフィー
ルド名と型の操作である。Cascading と Pig は共に１レコードをタプルで扱っている。
タプルとは、１つ以上の名前を持つフィールドで構成されるデータである。フィー
ルドは型の情報を保持しない。操作するフィールドの指定はフィールド名で行う。
型のキャストは自動に行われるが、誤った型の操作を行った場合、実行時にエラー
となる。解析アプリケーションの開発においても、フィールド名と型の操作を度々
行った。これらの操作はフィールド数が増えるにつれて煩雑になり、バグりやすかっ
た*1 。
*1
Pig では Zebra[8] を用いる事でフィールド名と型の情報をタプルに持たせた状態で操作できる。
パイプライン開発における Cascading と Pig の比較
2
パイプライン開発における Cascading と Pig の比較を行う。比較は次の三つの指
標で行った。
1. 拡張／再利用の容易さ パイプラインを構成するジョブの数が増えても生産性を落とす事なくパイプラ
インの拡張が可能であるか？コンポーネントを再利用可能な形で実装すること
が容易にできるか？
2. テストの容易さ 個々のコンポーネントを容易にテストできるか？テストするための機構が用意
されているか？
3. 開発サイクルの俊敏さ（Agility） 実装、テスト、実行、デバッグの開発サイクルを俊敏に繰り返す事ができる
か？そのための支援ツールは提供されているか？テスト用の小さいデータセッ
トからプロダクション環境の大きなデータセットでステップを踏んで実行でき
るか？
2.1
拡張／再利用の容易さの比較
Cascading では、Function, Filter, Aggregator, Buffer のインターフェースを実装
することで、フィールドの操作、タプルのフィルタリング、集計などの独自の処理
を再利用可能な形で実装できる。そして、Pipe クラスのビジネスロジックで独自実
装したクラスを使用する事ができる。Pipe クラスは入出力の指定をすれば、幾つで
も連ねることができるため、再利用が容易である。パイプラインの拡張においても
少数のパイプで小さく実装を開始し、パイプを付け足すことでより大きなパイプラ
インを構成できる。パイプラインが大きくなっても、バグ発生時の問題の切り分け
をパイプ単位で処理を実行することで迅速に行える。
一方、Pig は UDF を Java で実装することで再利用可能な処理を実装できる。し
かし、パイプラインの拡張においては、困難が伴う。ジョブの実行は PigLatin スク
リプトを通して行われるため、出力の分岐や処理のループといった複雑なジョブフ
ローになるとスクリプトを読み解くのが困難になる。また、Cascading のパイプの
ようなパイプラインの一部をモジュール化できるような機構がないため、ズラズラ
とスクリプトで処理を書かなければならない*2 。
以上より、拡張／再利用の容易さにおいては Cascading が Pig より優れていると
いえる。
*2
この課題に関しては、Pig Wiki[9] でも指摘されおり、改善案が提示されている。
2.2
テストの容易さの比較
Cascading でのテストは Junit の TestCase クラスを継承した CascadingTestCase
を用いて独自実装の Fuction, Filter, Aggregator, Buffer, Pipe のテストを行える。パ
イプラインのテストにおいても同様にコンポーネントの単位でテストができる。
Pig では Grunt のシェル上でインタラクティブにスクリプトの実行ができ、トラ
イ＆エラーの繰り返すことでテストを行える。Grunt のコマンドの illustrate(変数
の中身をルックアップ) や describe(変数のスキーマを表示)、explain(実行計画の表
示) を駆使する事で逐次、状態を確認できる (コード第 4 章.1)。PigUnit を使えば、
PigLatin スクリプトのユニットテストも行える。パイプラインのテスト場合、スク
リプトが長くなるため、Grunt で逐一、テストしたいの処理の箇所までスクリプト
を打ち込まないといけない。
UDF を使用した PigLatin スクリプトにバグがあった場合、UDF を Java で実装し
た箇所に問題があるのか？それとも、PigLatin スクリプトに問題があるのか？といっ
た Java と PigLatin の世界を行き来きしながら問題のきり分けをしなければならな
い [10]。
以上より、テストの容易さにおいては、Junit ベースのテストがシンプルにできる
Cascading が独自言語の PigLatin を用いる Pig より優れているといえる。
コード 第 4 章.1: Grunt のデバッグ用コマンド describe、illustrate、explain の実行例
1
2
3
4
5
6
7
8
grunt > records = load ’ excite - small . log ’ as ( sid : chararray , time : long , word :
chararray );
grunt > describe records
records : { sid : chararray , time : long , word : chararray }
grunt > illustrate records
2012 -02 -27 18:34:15 ,361 [ main ] INFO org . apache . pig . backend . hadoop . executionengine .
HExecutionEngine - Connecting to hadoop file system at : file :///
2012 -02 -27 18:34:15 ,381 [ main ] INFO org . apache . hadoop . mapreduce . lib . input .
FileInputFormat - Total input paths to process : 1
2012 -02 -27 18:34:15 ,381 [ main ] INFO org . apache . pig . backend . hadoop . executionengine .
util . MapRedUtil - Total input paths to process : 1
-------------------------------------------------------------------------------------
10
| records
| sid : bytearray
| time : bytearray | word : bytearray
|
-------------------------------------------------------------------------------------
11
|
9
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| B163FAFD64AFAB18 | 970916134159
| free downloadable pc wallpaper
|
-------------------------------------------------------------------------------------
---------------------------------------------------------------------------------| records
| sid : chararray
| time : long
| word : chararray
|
---------------------------------------------------------------------------------|
| B163FAFD64AFAB18 | 970916134159 | free downloadable pc wallpaper |
---------------------------------------------------------------------------------uster ’ stats - pig -0.0.1 - SNAPSHOT . jar ’;
grunt > define analyzeSession pig . AnalyzeSession ();
grunt > grouped = group records by sid ;
grunt > analyzed = foreach grouped {
>> sorted = order records by time ;
>> generate flatten ( analyzeSession ( sorted . time ));
>> };
grunt > explain
2012 -02 -27 18:38:51 ,331 [ main ] INFO org . apache . pig . backend . hadoop . executionengine .
HExecutionEngine - pig . usenewlogicalplan is set to true . New logical plan will
be used .
# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -# Logical Plan :
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
#----------------------------------------------fake : Store 1 -36 Schema : { pig . anal yzes essi on_ti me_3 :: record_count : long , pig .
ana lyze s ess i on_ t ime _ 3 :: session_time : long } Type : Unknown |
| - - - analyzed : ForEach 1 -35 Schema : { pig . an alyz eses si o n _ t i m e _ 3 :: record_count : long ,
pig . a naly ze s es s i on _ t im e_3 :: session_time : long } Type : bag
|
|
|
UserFunc 1 -30 function : pig . AnalyzeSession FieldSchema : pig .
analy z ese s sio n _ti m e_3 : bag ({ record_count : long , session_time : long }) Type :
bag
|
|
|
| - - - Project 1 -29 Projections : [1] Overloaded : false FieldSchema : time : bag ({
time : long }) Type : bag
|
Input : Project 1 -32 Projections : [*] Overloaded : false |
|
| - - - Project 1 -32 Projections : [*] Overloaded : false FieldSchema :
sorted : bag ({ sid : chararray , time : long , word : chararray }) Type : bag
|
Input : sorted : SORT 1 -33|
|
| - - - sorted : SORT 1 -33 Schema : { sid : chararray , time : long , word :
chararray } Type : bag
|
|
|
|
|
Project 1 -34 Projections : [1] Overloaded : false FieldSchema :
time : long Type : long
|
|
Input : Project 1 -31 Projections : [1] Overloaded : true
|
|
|
| - - - Project 1 -31 Projections : [1] Overloaded : true FieldSchema :
records : tuple ({ sid : chararray , time : long , word : chararray }) Type : tuple
|
Input : grouped : CoGroup 1 -21
|
| - - - grouped : CoGroup 1 -21 Schema : { group : chararray , records : { sid : chararray ,
time : long , word : chararray }} Type : bag
|
|
|
Project 1 -20 Projections : [0] Overloaded : false FieldSchema : sid :
chararray Type : chararray
|
Input : records : Load 1 -19
|
| - - - records : Load 1 -19 Schema : { sid : chararray , time : long , word : chararray }
Type : bag
52
53
54
55
56
#----------------------------------------------# New Logical Plan :
#----------------------------------------------... < 省 略 > ...
2.3
開発サイクルの俊敏さの比較
Cascading は Java の API のため、Java のアプリ開発と同じ環境で開発できる。
ローカルのファイルシステムをインプットとして受け取ることもできるため、小さ
なデータセットでローカルで試し、そして、疑似分散モード、プロダクション環境
での実行とステップを踏む事ができる。ただし、ジョブを実行するには jar の生成
が必要である。この開発サイクルは、生の MapReduce と同じで Cascading を用いる
事で、開発サイクルが俊敏にはなるわけではない。Cascading では、開発支援周り
の機能として、MR ジョブの実行計画を DOT ファイルに出力する事ができる。テス
トデータの自動生成やパフォーマンステストのツールは存在しない。
Pig は Grunt のシェルを用いる事で PigLatin スクリプトを書きながら、インタラク
ティブにジョブの実行が行える。local モードがあるため、小さなデータセットでロー
カルで試すこともできる。また、Grunt の変数の中身をルックアップできる illustrate
コマンド、変数のスキーマを表示する describe コマンドを用いることで素早くスク
リプトの挙動を確認できる。解析アプリケーションの PigLatin スクリプトの実装に
拡張／再利用の容易さ
テストの容易さ
開発サイクルの俊敏さ
○
△
○
△
△
○
Cascading
Pig
表. 2.1: Cascadig と Pig のパイプライン開発における比較（３段階評価）
おいても Grunt で繰り返し実行しながら修正を行った。Grunt は思いついたスクリ
プを手軽に実行できたため、俊敏な開発ができた。ただし、2.2 で述べた理由から
パイプラインが複雑になるにつれ、この俊敏さは失われると思われる。開発環境は、
PigLatin の eclipse のプラグインの PigEditor[11]、PigPen[12]、pig-eclipse[15] の利用
が可能である*3 。PigPen は、デバッグ用のデータの生成が行える。一方、PigEditor
はシンタックス/エラーハイライト、エイリアスの認識、オートコンプリートなどの
機能を提供する。他に開発支援ツールとしてパーフォーマンステストをするための
Pig Mix[13], Pig のモニタリングとデバッグを行う環境を独自に構築できるフレーム
ワーク Penny[14] がある。今回のレポートでは実際に試す事ができなかったが、Pig
Mix と Penny はドキュメントが少なく、使いこなせるまでの学習コストが高いよう
に感じられた。
以上より、開発サイクルの俊敏さの比較においては、Grunt のシェル環境がある
Pig が Cascading より優れているといえる。
2.4
比較結果
三つの指標を３段階で評価した結果を表 2.1 に示す。Cascading はモジュール化が
容易で「拡張／再利用の容易さ」と「テストの容易さ」に優れており、複雑なパイ
プラインの構築にも耐えうる実装が可能である。一方、Pig は Grunt というインタ
ラクティブに PigLatin スクリプトを実行できる環境があるため「開発サイクルの俊
敏さ」に優れており、シンプルなパイプラインの開発を素早く実装するのに適して
いる。
*3
PigEditor と PigPen のインストールを試みたが、Pig と eclipse のバージョンに制約があり、筆
者の環境では正常に動くまでに至らなかった。今回の開発では、シンタックスハイライトのみを行う
pig-eclipse を利用した。
第 5 章 結論・まとめ
本研究レポートでは、Hadoop 上でパイプラインジョブの実装を容易にするソフト
ウェア Cascading と Pig の比較を行った。比較を行うにあたり、検索ログの解析を
行うパイプラインを実装し、
「拡張／再利用の容易さ」
「テストの容易さ」
「開発サイ
クルの俊敏さ」の観点で比較を行った。比較の結果、Cascading は「拡張／再利用
の容易さ」と「テストの容易さ」に長けており、複雑なパイプラインの実装にも適
している。一方、Pig は「開発サイクルの俊敏さ」に長けており、Grunt 上でインタ
ラクティブにジョブを実行しながら、シンプルなパイプラインの実装に適している。
Hadoop Ecosystem という言葉があるように、Hadoop 関連のソフトウェアの開発
が盛んで、用途も多岐にわたる。そのため、どの開発にどのソフトウェアを適用す
ればよいのか？といったケーススタディに関するまとまった情報は決して多くはな
い。パイプラインの開発に絞ったレポートではあるが、業務でのソフトウェアの選
択の参考になれば幸いである。
参考文献
[1] manboubird/hadoop-stats - GitHub （検索ログの解析アプリケーション）,
https: // github. com/ manboubird/ hadoop-stats
[2] Building
Scale
Free
Applications
with
Hadoop
and
Cascading,
http: // www. slideshare. net/ cwensel/
building-scale-free-applications-with-hadoop-and-cascading-1616859
[3] Cascading, http: // en. wikipedia. org/ wiki/ Cascading
[4] Introducing Crunch: Easy MapReduce Pipelines for Hadoop, http: // www.
cloudera. com/ blog/ 2011/ 10/ introducing-crunch/
[5] Craig Chambers , Ashish Raniwala , Frances Perry , Stephen Adams , Robert
R. Henry , Robert Bradshaw , Nathan Weizenbaum, FlumeJava: easy, efficient
data-parallel pipelines, Proceedings of the 2010 ACM SIGPLAN conference on
Programming language design and implementation, June 05-10, 2010, Toronto,
Ontario, Canada
[6] Apache Hadoop:
What are the differences between Crunch
and
Cascading?,
http: // www. quora. com/ Apache-Hadoop/
What-are-the-differences-between-Crunch-and-Cascading
[7] DataFu - Hadoop library for large-scale data processing,
sna-projects. com/ datafu/
[8] Zebra and Pig,
html
http: //
http: // pig. apache. org/ docs/ r0. 9. 2/ zebra_ pig.
[9] Pig Wiki - Turning Complete Pig,
TuringCompletePig
http: // wiki. apache. org/ pig/
[10] Why Yieldbot chose Cascalog over Pig for Hadoop processing - BackType Technology, http: // tech. backtype. com/ 52456836
[11] PigEditor, http: // romainr. github. com/ PigEditor/
[12] PigPen, https: // issues. apache. org/ jira/ browse/ PIG-366
[13] Pig
Mix,
PigMix
https: // cwiki. apache. org/ confluence/ display/ PIG/
[14] Pig Penny,
Penny
https: // cwiki. apache. org/ confluence/ display/ PIG/
[15] pig-eclipse, http: // code. google. com/ p/ pig-eclipse/