データ解析用ワークフローフレームワーク Honey の紹介

最近，もっぱら上原ひろみさんの曲をエンドレスに聴いて癒しを得ています．もちろんピクルス作りも最高です．みなさんは何で癒しを得ていますでしょうか．こんにちは，技術部の石川有です．

以前，「mixi の解析基盤とApache Hive での JSON パーサの活用の紹介」で mixi における Hadoop/ Hive の活用の仕方について記事を書かせていただきました．今回の記事では，ちらっと触れていた Hive などで定期実行する必要のある処理をワークフローとして定義するフレームワークについて書きます．

本文章の構成

まず最初に，今回ご紹介するデータ解析用ワークフローフレームワーク Honey とは何か，なぜ作ったのかを説明します．つぎに，どのような構成や機能があるのかを簡単に説明します．それから具体的なデータ解析処理を記述する方法について説明します．その中で，定型的な処理を YAML として記述する方法を詳細に述べます．最後にこれから目指したいことを述べます．

データ解析用ワークフローフレームワーク Honey とは

なぜ Honey を作ったのか？
独自フレームワークを作るにあたって主に意識したこと

データ解析用ワークフローフレームワーク Honey の構成や機能

拡張しやすい構成を目指した
ワークフローの定義方法
処理の依存関係
その他の仕様

データ解析処理の記述方法

記述方法
YAML で処理を定義

将来的に目指したいこと

データ解析用ワークフローフレームワーク Honey とは

以前の記事の中で紹介したとおり，mixi では現在データ解析の主なツールとして Apache Hive を利用しています．Hive に限らず，データ解析をするためには，データを用意したり，どこかに転送したり，それを解析システムに取り込んだり，それからようやくデータ解析をして，整形をするという「流れ」があります．このような「流れ」，つまりワークフローを定義して処理が行えると管理がしやすくなります．

そこで社内の環境や事情にあわせて，独自のワークフローフレームワーク Honey を Perl で実装し，それを利用して Hive にクエリを投げるなど行なっています．

なぜ Honey を作ったのか？

Hadoop 関連のワークフロースケジューラというと，Apache OOzie や Azkaban などがあります．そのような中でも，独自のデータ解析用ワークフローフレームワークを実装するに至ったにはいくつか理由があります．

各エンジニアがより簡単に定期実行に必要なデータ解析ができるようになるシステムを提供したかった
サービスを Perl を実装しており，定数が Perl モジュール中にあるため集計結果のラベリング変換に Perl が必要であった
Hive から集計した結果をさらに細かく整形するための処理を加えたいなどより汎用的な利用ができると便利

独自フレームワークを作るにあたって主に意識したこと

処理単位の粒度を自由に定義できること
定期実行のデータ解析処理の各開発者のコストを減らすこと
将来的に Hive よりよいデータ解析用のシステムが登場したときに簡単に拡張できること
将来的な拡張を見据えたアーキテクチャ
手続きコストをできるだけ減らすこと
オレオレ実装をしないこと

なお，2012-11-15 時点では，Github などには公開しておりません．しかし将来的に，公開できるようにしたいと考えております．

データ解析用ワークフローフレームワーク Honey の機能紹介

拡張しやすい構成を目指した

前置きが長くなりましたが，上記で述べた意識したことをふまえてどのような構成になっているかを紹介していきます． 2012-11-16 時点で，このワークフローフレームワークでは Hive によるデータ解析処理を主に扱っています．しかし，将来的に Hive よりも利便性の高い集計システムが登場したときに，また同じような仕組みを構築するのはあまりにもコストが高いです．

そこでデータ解析用ワークフローフレームワークでは，３層構造をとっています．１番下の基本的なレイヤーは，ワークフローを定義したり，データ解析システムとして必要な機能を提供する層です．ここで言う必要な機能とは，ログを適切に出力することやアラートを適切に出すことなどです．この層についての要件は，後ほど詳細を説明します．つぎに真ん中のレイヤーでは，Hive などの処理を利用しやすくするレイヤーを用意しています．この層に Apache Pig や Cloudera Impala のような別のデータ解析システムを利用するための機能を追加して拡張することができます．最後に１番上のレイヤーでは，下の２層の利用して要求される個々の処理を記述します．各エンジニアは必要なデータ解析処理などを，個別に実装します．

ワークフローの定義方法

ワークフローフレームワーク Honey では，ChainOfResponsibility パターンを木構造として拡張し，個々のノードに処理を割り当てます．処理の親子の関係性は局所的に定義し，グローバルな管理を行なっていません．つまり個別に管理することで、処理の粒度を自由に選択することができます．

たとえば，日次でまとめて集計したい処理が下図のような木構造をとっていたとします．システム全体に障害が発生したとき，再集計のためのコマンド実行は木構造のルートの処理を指定すればよいです．それとは別に，ある特定の集計に問題が発生したとき，その問題のスコープの１つの集計を実行したいです．このような木構造の何処を指定しても，処理ができるような仕組みになっています．

仮にこのような処理をまとめて実行するのであれば，つぎのようなコマンドを１回投入します．このコマンドを実行することで上図赤線枠内の処理がすべて実行されます．--target に指定しているのは，木構造のルートの処理です．

worker.pl --target=Daily --mode=count --period=day --date=2012-11-16

しかし，あるサービスについてだけ（たとえば日記サービス）処理をやりたければ，つぎのようにコマンドを投入します．--target に指定しているのは，木構造のルートから１階層目の右側 Diary というノードを指しています．ルートノードである Daily と対象のノードの名前を指定することで，上図緑線枠内の処理がすべて実行されます．このようにして，処理の粒度を自由にコマンドラインから指定することができます．

worker.pl --target=Daily::Diary --mode=count  --period=day --date=2012-11-16

またこの木構造の各ノードには，複数の処理を mode として別々に割り当てることが可能です．たとえば，Hive にデータを載せたいときと削除したいときでは，それぞれのノードでの処理が分かれている必要があります．そして，木構造の関係も異なっていて欲しいです．そのような拡張に備えて自由に mode を増やせ，またその親子関係を別々に定義することができます．

このように必要な処理を記述さえすれば，柔軟に様々なデータ解析システムにも対応できます．またその処理の関係性を木構造で表現する事で，それぞれ分けた処理がどのように実行されるかを定義することができます．

処理の依存関係

つぎに，処理の依存関係について説明します．データ解析では，ある処理 A と別の処理 B が完了してから処理 C が実行されて欲しいときがあります．上で説明した木構造は，深さ優先探索で処理が実行されていきます．つまり，木構造下の「葉」から順次上のノードへと実行されていきます．深さ優先探索の性質を利用することで，たとえばあるデータのロードが完了したら，データ解析が開始されるというようなデータのロードと集計の実行のワークフローが定義できます．

その他の仕様

そもそも集計システムとして，どのような仕様が備わっているとよいかという紹介です．

１時間単位，日単位，週単位，月単位で処理単位を用意
期間や日次の指定

期間の制限
特定の曜日
月の特定の日
この日以降から処理を行う
この日以降は処理行わない

定期的なログのレポートの送信
重要な処理については即座にアラートを送信

ここまでは，ミクシィで利用しているデータ解析を定期実行で行うためのワークフローフレームワークを実装した動機や概念と仕様について紹介でした．ここからは，簡単にフレームワークを利用・拡張するための仕組みについて説明します．

データ解析処理の記述方法

最初に記述の種類について述べ，それから定型的な処理を YAML で定義する方法について詳細に述べます．ここで言う定型処理とは，たとえばダンプしたデータを Hive にロードするとか，Hive で実行したクエリを CSV として保存するなどの再利用性の高い処理のことです．

記述方法

Honey で処理を記述を定義する方法は，３つのルートがあります．本当に定型的な処理については YAML で処理を記述できるようにしています．定型的な処理ではすまないときは，対応する Perl モジュールを上記で示した ChainOfResponsibility パターンの１つのノードを実装することで対応できます．どうしても汎用的定型的な処理で記述できないときは，あきらめてそれぞれ実装を行ないます．逆に定型的な処理で済むものについては，必要な処理を TemplateMethod として共通的に準備しておくことで各開発者のコストを下げられるようにしています．

Perl のクラスとして必要な処理を実装
Perl のクラスと定型的な処理は対応する YAML に定義した情報から付与
定型的な処理しか必要ないときは，対応する YAML から Perl クラスを生成

定型処理を YAML で定義

定型的な処理については，YAML のみで記述することができます．この YAML の値を元に ChainOfResponsibirity パターンの１つのノードクラスが生成されます．たとえば，Hive で実行したクエリの結果を CSV ファイルに保存するときは，つぎのような YAML ファイルになります．

保存されるCSVのファイル名は，diary.uu
CSV のヘッダは，日付とUU
依存関係として，Diary の load という処理が完了してから処理が実行
期間の起算は対象の日付け
集計の単位は，日と週のみ
月曜日のみ実行
週の実行のときは，指定した日から１週間さかのぼった日付けまでさかのぼった期間


---
TYPE: HandlerDefinition
DECORATORS:
    SetAccessors:
    CSV_FILE_NAME: 'diary.uu'
    CSV_HEADER: ['日付', 'UU']
    children:
        count: 
        - handler: Diary
          mode: load
    validator:
        - ValidateDayCommand:
            day_of_week: 1
        - ValidatePeriodCommand:
            count: ['day', 'week']
SetTerm:
    is_base: 1
SetMethods:
    count:
        template: 'Honey::Builder::Template::Common::SaveHiveResultAsCSV'
        params:
            query: |-
                USE event;
                SELECT
                    '__START__',
                    count(distinct member_id)
                FROM diary_create
                WHERE __PERIOD_CONDITION__;

この処理を週の処理として実行すると，つぎのようなクエリが Hive で実行されます．この処理を Diary::Uu という処理として，対応するような YAML にしてあります．処理の実行日を 2012-11-19 にすると，上で説明した通り 2012-11-19 から起算した一週間分になります．


worker.pl --target=Diary::Uu --mode=count --date=2012-11-19

USE event;
SELECT
  '2012-11-13',
  count(distinct member_id)
FROM diary_create
WHERE dt IN ('2012-11-13', '2012-11-14', '2012-11-15', '2012-11-16', '2012-11-17', '2012-11-18', '2012-11-19');

このようにして，YAML にしていした TemplateMethod を利用して必要な定型処理を記述することができます．