BLOGSpeee開発陣による技術情報発信ブログです。メディア開発・運用、スマートフォンアプリ開発、Webマーケティング、アドテクなどで培った技術ノウハウを発信していきます!2016-07-07AirflowによるデータパイプラインのスケジュールとモニタリングTweetこんにちは!WEBマーケティング事業部エンジニアの@hatappiです。現在私のプロジェクトでデータパイプラインのスケジュールとモニタリングをAirflowというオープンソースのツールを使用して運用しています。今回は導入や使い勝手の話です。Airflowとは?airbnbがオープンソースで開発したワークフロー管理プラットフォームです。Incubatorのオープンソースプロジェクトになっています。Airflowはバッチ同士の関係を管理し、可視化することが出来ます。実行した際のLogや実行時間の推移など、様々なデータの閲覧をデフォルトの機能として使用することが出来ます。データパイプラインとは?プロジェクトを運用していると、複数のテーブルにデータが入っており、それらのサマリーを作成するといった経験をしたことはありませんか?これらの一連のプロセスのことをデータパイプラインといいます。私のプロジェクトではMySQLやEmbulkを用いてRedshiftにバルクインサートを行い、バッチ処理をして最終的なデータにするデータパイプラインがあります。cronではないでしょうか。複数のスケジュールが追加された際に管理がし辛い。ログの出力設定をしなければログが出力されず、問題が起きた際の原因特定が難しい。動いているタスクの進捗を確認するためには、ログの確認やタスクの中に進捗を通知するような仕組みを作らなければならないといった点に注意しなければなりません。弊社では昔は定期実行処理をJenkinsで行っていました。記事はこちらしかし今回はスケジューリングとモニタリングをしたいだけで、Jenkinsほど多機能なツールは必要なかったため、スケジュールとモニタリングに特化したAirflowを使うことにしました。インストールについてpipを使ってAirflow本体をインストールします。airflowAirflowは本体とは別にサブパッケージが用意されており、MySQLやHive、S3やSambaなど様々なパッケージが提供されておりそれらを使用することでデータ連携を行うことが出来ます。今回はAirflow上のデータストアとしてMySQL、ジョブキューとしてCeleryを使用します。これらのサブパッケージも本体と同様にpipを使ってインストールします。airflow[celery]Airflowでは、データストアをどこに置くか、ジョブキューは何を使うかなどをすべて設定ファイルで管理します。今回はデータストアとジョブキュー部分を記載しています。他にも設定できる項目はあるので詳しくは公式ドキュメントを見てください。db+mysql://[user_name]:[password]@[host]:[port]/[db_name]これで一通りの設定が完了しました。下記コマンドを実行するとブラウザから管理画面を確認できるようになります。8080Airflowの構成AirflowではDAGとよばれる複数のタスクの集まりを1つのスケジュールとして扱い管理をしています。DAGの詳細についてはwikipediaをご覧ください定義されたDAGは下記の3つの役割をもったプロセスを立ち上げることで運用をはじめることが出来ます。ここではそれぞれの役割を簡単に記載します。次に1つ簡単なDAGを定義してみます。v1.7.1.2今回は例として毎日16:00になるとdateコマンドを実行するDAGを定義します。dag=dag)task1.set_downstream(task2)プログラム内にstart_dateがありますが、ここではいつからタスクを開始するかを指定します。ここで気をつけなければならないのが、datetime.now()などと指定しないことです。指定を間違えてしまうと延々とタスクが走り続けてしまいます。00)}下記でDAGインスタンスを定義します。schedule_intervalにはスケジュールを記載します。例では馴染みのあるcronの記法で指定しています。task1の後にtask2を実行するというDAGを定義できます。task1.set_downstream(task2)Pool運用していると出てくるのが、複数のタスクを並列して走らせる要望です。AirflowではPoolという機能があり、こちらを設定することで実現することが出来ます。並列化に使うDAGを定義します。dag=dag)task1.set_downstream(task2)task1.set_downstream(task3)task2.set_downstream(task4)task3.set_downstream(task4)今回はこのprint_hello-worldとprint_hello-japanを並列実行させたいと思います。Poolと遷移します。/admin/pool/)こちらでPool名と並列実行数であるslotsを定義します。先ほど定義したPool名を使用してDAGを編集します。これでslotsで定義した数だけ並列実行が出来るようになります。dag=dag)Slackとの連携についてタスクの実行結果を把握しやすくするために、Slack連携を行います。airflowにはサブモジュールでairflow[slack]がありますが、これはタスクの一つとして実行されるものです。今回使用したいのは、それぞれのタスクが完了もしくは失敗した際にslackに通知するものです。これが実現できると随時DAGの進捗を受け取ることが出来るので、管理画面を見に行く必要がなくなります。これを実現するのがon_success_callbackとon_failure_callbackです。これらのコールバックにSlack通知を送る関数をセットすることで、通知を行うことができます。webhookを使うため、こちらでhookのurlを取得します。slackwebでインストールしておきます。下記が実際に成功時、エラー時にSlackに通知を送るコードです。slack.notify(attachments=attachments)後はコールバックにこの関数を指定しておけば、タスクの実行結果がSlackに通知されるようになります。notify_error,}成功した場合は緑、失敗した場合は赤になるよう設定したので、通知の結果は下記の画像のようになります。実際に使ってみて実際に運用していくと下記のようなグラフやガントチャートが表示されます。これを見ることで、通常よりも時間がかかっている処理を発見できたり、問題のある箇所のログを確認することができるようになり、原因究明がしやすくなりました。どのスケジューラにも言えることですが、ツールにロジックをもたせ過ぎないことが大切です。今回はPythonでタスクを定義しましたが、実装次第ではDBから取得した値を用いて、タスクに渡す引数を変更することも可能です。これを行ってしまうと、データとタスクが密結合になってしまい修正をする際にスケジュールも変更しなければならないなど運用コストが上がってしまいます。デバッグもしやすくなりました。大きな利点としては下記のようにエンジニア以外のメンバーもタスクを実行できるようになるので、エンジニアは開発に専念できるようになります。運用観点で掘り下げた記事を書きます。