memo: Airflowの紹介 #wfemeetup

• SIer
• Bigtop/Yetusコミッタ

アジェンダ

• 背景
• 紹介
• デモ
• 要望

背景

• 以前はOozieを使っていた
  • 設定はXMLで定義するのでだるい
  • fork-joinを使うためにはすべての経路が成功する必要がある
  • どれかがOKだったら全部OKみたいな運用ができない
  • 任意のDAGとして定義できない
  • coordinatorを使う場合はデータが到着してることが前提
  • UIが古い
• Airflowに可能性を感じた
  • ワークフロー Python
  • Trigger Rules
  • フローに制約がない

紹介

• 2016/03にAirbnbがApacheに寄贈したやつ
• 1.7(AIRFLOW-921配下のissueが全部解決されたらリリースされると思う)
• pip install airflow
  • 導入の敷居は高くないかも

中心的な概念

• DAG
  • 全体のタスク
• Operator
  • 定義
• Task
  • 処理の各ジョブ
  • インスタンス
• Task Instance
  • 実行結果
  • ステータスの保持
  • ID以外にタイムスタンプを持っている

Operatorの例

• コマンド発行
  • BashOperator DokcerOperator
• SQL発行
  • HiveOperator
  • JdbcOperator
  • MySQLOperator
  • PigOperator
• データ転送
  • S3
• 通知
  • Email
  • HTTP
  • Slack
• Operatorの独自定義も可能

その他の概念/機能

• connection: データストアの接続管理(パスワードなど)
• hook: データのload/dump
• pool: タスクの並列数
• queue: celery 外部のキューイングシステムをジョブキューとしてつかえる
• branching: DAG中の条件分岐を実現
• SLA: 一定時間内に成功しなかったやつを通知

デモ

• ペットショップの売上データの整形集計分析
  • bigtopのサンプル https://github.com/apache/bigtop/tree/master/bigtop-bigpetstore/bigpetstore-spark
  • 売上データの生成→クレンジング→売上集計→顧客ごとの商品レコメンド
• スケジューリングはcronの記法 or Python記述 @onceで一度きり
• generator/etl/stats/recommend

不足している機能

• HA(high availability)構成のサポート
• 運用性の工場、DAGの登録/更新/削除をweb uiから実施できるようにしたい
• タイムゾーンがUTCでしかサポートしていない

• 任意の場所で処理を停止/再開したい

• Airflow Meetup Tokyoをやろう

• 西海岸では数カ月に１回くらいやってるぽい
• パッチを投げてほしい