と Hadoopによるデータ解析

doryokujin

Tokyo Linux Study 02 #tlstudy (2011/07/13)

[名前] doryokujin ( 井上 敬浩 )

[年齢] 26歳

[専攻] 数学（統計・確率的アルゴリズム）

[会社] 芸者東京エンターテインメント

[職業] データマイニングエンジニア

[趣味] マラソン ( 42.195km: 2時間33分 )

[コミュニティ]

・MongoDB JP・TokyoWebMining ・おしゃれStatistics・分散処理ワークショップ

自己紹介

1. Simple Daily Batch Processing Model

・Hadoop --> MongoDB

・MongoDB As “Result Data Storage”

2. Flexible Hourly Batch Processing Model

・MongoDB --> Map Reduce --> MongoDB

・MongoDB As “Raw Data Storage” && “Data Processing Engine” && “Result Data Storage”

本日の内容

0. What is MongoDB ?

MongoDB System (mongod, mongos)

Database

Document

Document

Document

Document

Document

Document

Document

Document

Collection

MongoDB データモデル

Database

CollectionCollection Collection

{

"_id" : ObjectId("4dcd3ebc9278000000005158"),

"timestamp" : ISODate("2011-05-13T14:22:46.777Z"),

"binary" : BinData(0,""),

"string" : "abc",

"number" : 3,

"subobj" : {"subA": 1, "subB": 2 },

"array" : [1, 2, 3],

"dbref" : [_id1, _id2, _id3]

}

padding

{

"_id" : ObjectId("4dcd3ebc9278000000005158"),

"timestamp" : ISODate("2011-05-13T14:22:46.777Z"),

"binary" : BinData(0,""),

"string" : "abc",

"number" : 3,

"subobj" : {"subA": 1, "subB": 2 },

"array" : [1, 2, 3],

"dbref" : [_id1, _id2, _id3]

}

db.coll.find({"number": 3});

db.coll.find({"number": {$gt: 1}});

db.coll.find({"subobj.subA": 1});

db.coll.find({"subobj.subB": {$exists: true} });

db.coll.find({"string": "abc"});

db.coll.find({ "string" : /^a.*$/i });

padding

db.coll.find({"array": {$all:[1, 2]} });

db.coll.find({"array": {$in:[2, 4, 6]} });

Query

{

"_id" : ObjectId("4dcd3ebc9278000000005158"),

"timestamp" : ISODate("2011-05-13T14:22:46.777Z"),

"binary" : BinData(0,""),

"string" : "def",

"number" : 4,

"subobj" : {"subA": 1, "subB": 2 },

"array" : [1, 2, 3, 4, 5, 6],

"dbref" : [_id1, _id2, _id3]

"newkey" : "In-place"

}

{ $inc : {"number": 1} }

{ $set : {"string": "def"} }

{ $pull : {"subobj": {"subB": 2 } } }

{ $addToSet : { "array" : { $each : [ 4 , 5 , 6 ] } } }

{ $set : {"newkey": "In-place"} }

Update

1. Simple Daily Batch

Processing Model Source --> Hadoop --> Mongo --> WebUI

解析対象とゴール

Input: 行動ログuserId

カオス

Keyの種類によっては配列やさらに入れ子になったオブジェクトも

_id は”日付”+”userId”+”行動タイプ”の独自のStr型ユニークキー(後に

Objectidに変更)

日付・userId・行動タイプなど、様々な軸で集計可能

Output: MongoDB

userId

1. 分散するサーバーから各種ログをローカルに収集2. 課金・登録情報MySQLから該当データを取得3. ユーザーゲームセーブデータをCassandraから取得

1. 解析フロー

1.Data Gathering

3.Data Analysis

5.Data Mining

6.Data Sharing

2.Data Pre-processing

4.Reslt Data Strage

1. ログの整形2. フィルタリング

1. ユーザーIDをキーにした各指標値の集計2. アクセスログの集計3. 一定期間のイベントの効果測定のための集計

ランキング課金情報登録日

行動ログアクセスログ

ゲームセーブデータ

2. Data Pre-processing3. Data Analysis 4. ResultData

Storage

Python ScriptThrift & Python

データセンター → AmazonS3

1. Data Gathering

Cassandra MySQL

HDFS

1. Data Gathering

5. Data Sharing6. Data Mining

1. 解析フロー

exec/day

Sleepy.Mongoose

解析対象データ

ソーシャルデータ

Graph Analysis

Web UI

Data Analysis

5. Data Sharing

6. Data Mining

ScientificPython

1. 解析フロー

[構成]

[1] MongoDB Result Server × 2 : Result Data Storage

・SIngle Server + Master / Slave 構成

[2] Hadoop Server × 3 : MapReduce

1. 技術要素・構成サーバー

Slave2

Master1

Slave1

Master2Replication

Input: 行動ログ --> Hadoop

ランキング課金情報登録日

行動ログアクセスログ

ゲームセーブデータ

2. Data Pre-processing3. Data Analysis 4. ResultData

Storage

Python ScriptThrift & Python

データセンター → AmazonS3

1. Data Gathering

Cassandra MySQL

HDFS

1. Data Gathering

5. Data Sharing6. Data Mining

exec/day

1. 解析フロー

・Python + Hadoop = Flying Circus Elephant

・Last.fm で使用

・Hadoop Streaming を実行

・map/reduce の記述が楽

・簡単な join もできる

・デバッグが容易

Dumbo: Flying Circus Elephant

dumbo start wordcount.py \

-hadoop /path/to/hadoop \

-input wc_input.txt \

-output wc_output

def mapper(key, value):

for word in value.split(): yield word,1

def reducer(key, values):

yield key,sum(values)

if __name__ == "__main__":

import dumbo

dumbo.run(mapper, reducer)「wordcount.py」

Dumbo で Word Count

[1st MR]

・Map で各入力から Key(入れ子) と Value を計算

・Combine で Key ごとの合計を計算 (Mini-Reduce)

・Reduce Key ごとに合計を計算 -> HDFS上に保存

[2nd MR]

・複数のMapper からMongoDBに並列書き込み

・Reduce で統計情報を計算、出力

1. Hadoop --> MongoDB

その後圧縮してローカルストレージに保存

Keyの種類によっては配列やさらに入れ子になったオブジェクトも

_id は”日付”+”userId”+”行動タイプ”の独自のStr型ユニークキー

日付・userId・行動タイプなど、様々な軸で集計可能

1. Hadoop--> MongoDB

1. MongoDB --> WebUI

JSONSleepy.Mongoose

・MongoDB <--> jQuery は JSON を介してシームレスにやりとりが可能

・Web Front <--> MongoDB はRestインターフェースであるSleepy.Mongooseやnode.jsライブラリで

1. MongoDB --> WebUI

1. MongoDB --> WebUI

・メモリ大量消費：たくさんのフィールドにインデックスを作成したため、（しかも_idはString型で非効率）

・ディスク不足（1000万 write/ day）

・ネットワーク帯域圧迫：デイリーのログの転送に非常に時間がかかる

・小規模の非バッチ処理が増えていちいちHadoop使うのがしんどい、もっとシンプルに！カジュアルに！

・生ログ自身にアクセスできる環境が必要になってきた

1. 問題点

・メモリ大量消費：たくさんのフィールドにインデックスを作成したため、（しかも_idはString型で非効率）

・ディスク不足（1000万 write/ day）

・ネットワーク帯域圧迫：デイリーのログの転送に非常に時間がかかる

・小規模の非バッチ処理が増えていちいちHadoop使うのがしんどい、もっとシンプルに！カジュアルに！

・生ログ自身にアクセスできる環境が必要になってきた

1. 問題点 上記2点はスケールアウトにて対応

Daily -> Hourly 単位の処理へ切り替え

軽量集計フレームワークが必要

生ログをストレージする仕組みが必要

2. Flexible Hourly Batch

Processing ModelSource --> MongoDB --> MapReduce

--> MongoDB --> WebUI

[エンジニア向け]・問題が起こった場合、該当する時間のログを提供して早期問題解決に役立てる

・不正値や異常値を発見して不正ユーザーの対策や早期バグの発見に役立てる

[サポート向け]・ユーザーの問い合わせ・クレームに対して根拠をもって返答し、納得してもらえるように、対象ユーザーの行動を全てトラッキングできるようにする

2. ログストレージの必要性

・ログの表示機能：

指定された期間のログを提供する機能

・ログの検索機能：

ユーザーIDや行動タイプ、期間の条件でログを切出す機能

・バッチ集計機能：Daily --> Hourly

ユーザーID・日・時間ごとにあらゆる指標を計算する機能

・非バッチ集計機能： Hadoop --> More Easy!!

急な要請にも即座に該当データの提示や集計ができる機能

2. 解析基盤に求められるもの

2. 解析フロー1.Data Gathering

4.Data Analysis6.Data Mining

7.Data Sharing

2.Data Pre-processing

5.ResltData Strage

3. Raw Data Strage

Raw Data Search

User Tracking

4. ResultData Storage

ランキング課金情報登録日

ゲームセーブデータ

Python ScriptThrift & Python

Cassandra MySQL

1. Data Gathering

5. Data Sharing6. Data Mining

2. 解析フロー

行動ログアクセスログ

1. Data Gathering2. Data Pre-processing

データセンター

insert/hour

3. Data AnalysisMongo Map Reduceor Hadoop

[構成]

[1] MongoDB Server × 12： Raw Data Storage

[2] MongoDB Server × 4～ : Result Data Storage

[3] Hadoop Server × 17 : Map Reduce（[1]と共存、15 datanode: over 50 mapper）

[4] Other Server : Monitor Server, Test Server, etc...

2. 技術要素・構成サーバーReplica Set:

6 Primary + 6 Secondary

[25 shard 構成]

・25 Shard × 3 Member Replica Sets + 3 config + 6 mongos

・Shard key = “hour”

・shard00 - shard23 に各時間 (0,1,...,23) を割当るマニュアル Sharding、Chunk の分割も起こさない

・自動 Balancer の停止：Migration は起こさない

2. Raw Data --> MongoDBManual Sharding Model

Sharding の種々の問題を防ぐ

[2011-07-01 12:01:48,447]

[4Shard / Server]

・1Core / Shard の制約

・サーバー1台 (4core) に4shard=(4mongod)を同居

[Output Only Shard]

・25 番目の shard は解析結果コレクションの書き出し用

・ログデータは保持しない

・SSDによる高速出力

[Replica Set]

・2mongod + 1arbiter のReplica Set を構成

全コア活用

高速書き出し

2. Raw Data --> MongoDB

Source --> MongoDB

Shard00

Shard01

Shard02

Shard03

Shard04

Shard05

Shard06

Shard07

Shard04

Shard05

Shard06

Shard07

Shard20

Shard21

Shard22

Shard23

Shard20

Shard21

Shard22

Shard23

Shard24 SSD

mongos

Primary

Secondary

node1

Shard00

Shard01

Shard02

Shard03

node2

node6

node7 node8 node12

Client

Replication Replication Replication

Shard01には01時のデータが入るよう

にShardingPyMongo

構成

Shard00

Shard01

Shard02

Shard03

Shard04

Shard05

Shard06

Shard07

Shard04

Shard05

Shard06

Shard07

Shard20

Shard21

Shard22

Shard23

Shard20

Shard21

Shard22

Shard23

Shard24 SSD

mongos

Primary

Secondary

node1

Shard00

Shard01

Shard02

Shard03

node2

node6

node7 node8 node12

Client

Replication Replication Replication

mongosが該当するshardからデータ取得

query

Read

Shard00

Shard01

Shard02

Shard03

Shard04

Shard05

Shard06

Shard07

Shard04

Shard05

Shard06

Shard07

Shard20

Shard21

Shard22

Shard23

Shard20

Shard21

Shard22

Shard23

Shard24 SSD

mongos

Primary

Secondary

node1

Shard00

Shard01

Shard02

Shard03

node2

node6

node7 node8 node12Replication Replication Replication

insert / hour

db.collection.insert( {hour:0, userId:”1234”, actionType:”login”,});

mongosが該当するshardへデータ挿入

PyMongo 行動ログStreaming...Write

Mongo --> MapReduce

m = function(){

this.tags.forEach{

function(z) {

emit(z, {count: 1});

}

};

};

r = function(key, values) {

var total=0;

for (i=0, i<values.length, i++)

total += values[i].count;

return { count : total };

}

res=db.things.mapReduce(m,!r);

# クエリによるフィルタリングや finalize 処理も可能

Mongo Map Reduce:JavaScriptベース

Mongo Map Reduceで Word Count

Mongo Shell からカジュアルに実行

N-Mapper-Combiner 1-Reduce“Primary”に

指定した Shard

ResultCollection

out option1. replace2. merge3. reduce(Inclement)4. memory

3. Reduce

※ N = Shard 数

※ Combiner = Reducer

※ Shuffle 無し

mongos1.ルーティング・指示

Shard1

2. Map+ Combine

Shard2 Shard3 Shard4 Shard5

Shard6

Shard1

2. Map+ Combine

2. Map+ Combine

2. Map+ Combine

2. Map+ Combine

2. Map+ Combiner

MR Model

Shard00

Shard01

Shard02

Shard03

Shard20

Shard21

Shard22

Shard23

SSD

Mongo Map Reduce (for small data)

Shard00

Shard01

Shard02

Shard03

24shard(=24core) を活用して

MongoMapReduce SSD上のcollectionに高速書き込み

mongos

2.MR 2.MR 2.MR2.MR 2.MR

1.ルーティング・指示

．．．

ResultCollection

Shard24

3. Reduce

out:reduce

node1 node2 node6

・単純和集計モデルなのでMongo MRでも処理が書ける

・性能遅（Secondary非活用・スレッド制限等）

・→重い処理はRead/Writeの他のオペレーションを明らかに妨害する（ロックはしない）

・前述した行動ログはカオス

・→最も面倒なのは各入力からKeyとValueを特定し、抽出すること

・i.e. キーごとの細かいテキスト処理が重要

2. 感想

・単純和集計モデルなのでMongo MRでも処理が書ける

・性能遅（Secondary非活用・スレッド制限等）

・→重い処理はRead/Writeの他のオペレーションを明らかに妨害する（ロックはしない）

・前述した行動ログはカオス

・→最も面倒なのは各入力からKeyとValueを特定し、抽出すること

・i.e. キーごとの細かいテキスト処理が重要

2. 感想MongoDBの外部で大規模集計を実行

JavaScriptでは不自由

柔軟な処理の記述が必要

もっとパワーを！

Shard00

Shard01

Shard02

Shard03

Shard04

Shard05

Shard06

Shard07

Shard04

Shard05

Shard06

Shard07

Shard20

Shard21

Shard22

Shard23

Shard20

Shard21

Shard22

Shard23

Primary

Secondary

node1

Shard00

Shard01

Shard02

Shard03

node2 node6

node7 node8 node12

Mongo Hadoop Model (for big data)

github: mongodb / mongo-hadoop

・Mongo-Hadoop Plugin (from 10gen)

- Chunk や Shard の数だけ mapper を立ち上げ可能

- Secondary からもインプット対応（と思われる）

- さらに Pig 対応も進んでいる

- Flume の sink に MongoDB を指定可能

github: mongodb / mongo-hadoop

Mongo - Hadoop

Pydoop: Hadoop API for Python

MotivationArchitecture

ExamplesConclusions and Future Work

Third Party Solutions

Hadoop-based: same limitations as Streaming (Dumbo) andJython (Happy), except for ease of useOther implementations: good if you have your own cluster

Hadoop is the most widespread implementation

Leo, Zanetti Pydoop: a Python MapReduce and HDFS API for Hadoop

Motivation

Architecture

Examples

Conclusions and Future Work

Summary of Features

Streaming Jython Pydoop

C/C++ Ext Yes No Yes

Standard Lib Full Partial Full

MR API No* Full Partial

Java-like FW No Yes Yes

HDFS No Yes Yes

(*) you can only write the map and reduce parts as executable scripts.

Leo, Zanetti Pydoop: a Python MapReduce and HDFS API for Hadoop

・PythonベースのHadoopライブラリはDumboを初め、いくつかある

・PydoopはHDFSの操作も行え、独自のInputFormat等も記述可能

DisCoも大変興味深い

Python MapReduce Programming with Pydoop

Pydoop: Hadoop PipesMotivation

Architecture

Examples

Conclusions and Future Work

Hadoop Pipes

Communication with Java

framework via persistent

sockets

The C++ app provides a

factory used by the framework

to create MR components

Providing Mapper and

Reducer is mandatory

Leo, Zanetti Pydoop: a Python MapReduce and HDFS API for HadoopPython MapReduce Programming with Pydoop

PydoopMotivation

Architecture

Examples

Conclusions and Future Work

Integration of Pydoop with C++

Integration with Pipes:

Method calls flow from the

framework through the C++ and the

Pydoop API, ultimately reaching

user-defined methods

Results are wrapped by Boost and

returned to the framework

Integration with HDFS:

Function calls initiated by Pydoop

Results wrapped and returned as

Python objects to the app

Leo, Zanetti Pydoop: a Python MapReduce and HDFS API for Hadoop

Python MapReduce Programming with Pydoop

・PydoopによってHDFSの操作など、Dumboよりも処理の自由度が大幅にアップした

・Numpy などのライブラリで複雑な処理も可能

・ただし前述したKey、Valueの記述は面倒

・Hadoopのデバッグも面倒、もっとカジュアルに！！

・Data Streaming (Scribe) Model の問題：

・中継サーバーダウンによるデータロストの心配

・Mongoへの書き込みが途中で失敗した場合のロールバック問題

・その他多数（詳細はまたの機会にでも…）

2. 感想

・PydoopによってHDFSの操作など、Dumboよりも処理の自由度が大幅にアップした

・Numpy などのライブラリで複雑な処理も可能

・ただし前述したKey、Valueの記述は面倒

・Hadoopのデバッグも面倒、もっとカジュアルに！！

・Data Streaming (Scribe) Model の問題：

・中継サーバーダウンによるデータロストの心配

・Mongoへの書き込みが途中で失敗した場合のロールバック問題

・その他多数（詳細はまたの機会にでも…）

2. 感想 小規模向け・非バッチ処理用にHadoopよりももっと軽量で便利なフレームワークがあってもいい

Mapper

Mapper

Mapper

Mapper

Mapper

Mapper

Mapper

Mapper

Mapper

Pipelined MR Model (for middle data)

①②

③

① タスクキューにタスクが入り次第、タスクを随時取得し、並列実行(Online MR)② 各Shardでデータ取得 && シェルスクリプトベースでの Map-Combine

③ 集計の終わったタスクから順次ReduceへタスクをPush、Redisへインクリメント④ Daily,HourlyのタイミングでGlusterFS へデータ書き出し→MongoDBへ書き込み

④ ④④

①

Mapper

Mapper

Mapper

②

②

②

②

②

③

③task queue

Block 情報を含むタスクを定期的を追加

Input Source

SSD SSD SSD

gawk '

BEGIN{ reducenum='$REDUCE_NUM'; }

{ userid=$7; key=$8; }

key ~ /a\{GetLoginBonus\}/ { incrby(userid,key,$9,a); next;}

key ~ /a\{SideJob\}/ { incrby(userid,key,$11,a); next;}

key ~ /a\{CleanMyShop\}/ { hincr(userid,key,$9,a); next; }

key ~ /(GetAvatarPart|ChangeP|ChangeWakuwakuP|ChangeKonergy)/

{ incrbydiff(userid,key,$9,a); next; }

...‘ $IN

# for reducer1 (such as “userid % reducenum == 0”)

# command userid key value

MULTI

HINCRBY 1111 a{ChangeGreed} 3

HINCRBY 1111 a{GianEvent} 7

HINCRBY 1111 a{TeamChallenge} 5

HINCRBY 2222 a{Battle} 3

HINCRBY 2222 a{ChangeMoney} 3

...

EXEC

各Map-Combine処理はawkで記述

Partitionされた(Key,Value)は各々のReduce先のRedisへ逐次集約される

HINCRBYコマンドで、key:1111のハッシュ型の key: ”a{GianEvent}”の

value を3インクリメント

・Shell Script ベースのMap Reduceフレームワーク自作

・In-memory 集計モデル(With Redis)。書き込みはSSD

・カジュアルなテキスト処理、集計

・32 Mapper - 6 Reduce 構成

・単純な和集計なので、Sort もMap/Reduce間のWaitも不要（Pipelined Map Reduce）

・高速でデバックも容易

・MongoDB、Hadoop、Original、状況に応じて使い分け

2. 感想

・MongoDB は Data Storage としての役割と Data Processing Engine としての役割を兼ねることが出来る

・集計・解析の他に生ログをストレージする必要がある場合はMongoDBは後の検索性も考慮すると最適

・しかし小規模や単純な和集計の場合はHadoopの制約に縛られない柔軟な計算フレームワークを作った方が良いかも

・入力をKey-Valueに分解することができればほぼ仕事は終わり。何をKey-Valueにするのか、試行錯誤しながら見つけるのが大変。そのために生データとしっかり向合うのが大事

まとめ

ありがとうございました

MongoDB:Production Deplyments