apache airflowとは
- https://github.com/apache/incubator-airflow
- pythonでタスクとタスク依存関係を書く
- プラグインによる拡張が可能
- ワークフロー管理のためのUI
- ワークフローの可視化
- 実行の監視、ログの確認(原因究明が速い)
- 必要なタスクの再実行
apache airflowの特徴
- pythonコードなので、ソースはgitで管理できる
- 処理を実行するworker(Scaling Out with Celery)の分散が可能
- 色々プラグインが用意されている
- DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperatorなど
- gcp(storage, bigquery)が直接に使える
- その以外は https://github.com/apache/incubator-airflow/tree/master/airflow/contrib
- pythonによるプラグイン拡張ができる
apache airflowの特徴
- 一つworkflowは一つDAG(Directed Acyclic Graph)
- タスクの集合及び依存関係
- 有向非巡回グラフ, cycleを持たない有向グラフのこと
- タスクの依存関係はトポロジカルソート(topological sort)で解決できる
- タスク間にデータの受け渡しは(‘xcom’ )
- “pull” data from a second task (xcom_pull)
- “push” data from one task (xcom_push)
類似ツール(workflow engine)
- spotify luigi(python)
- treasuredata digdag(java)
- oozie
GCP用のプラグインが多いので、airflowを選んだ
Airflow Daemons
- web server
- scheduler
- worker
Single Node Deployment:
Cluster Deployment
- workerがスケールできる
- web serverもスケールできる
- schedulerがスケールできない (single failure point)
- airflow-scheduler-failover-controller(active-standby)のプロジェクトがある
airflowスクリプト(dag)の定義
スクリプト(dag)内部のタスクとタスクの依頼関係を実現したい
タスク=Operator
airflowタスク(dag)の定義
import airflow
from builtins import range
from airflow.operators.bash_operator import BashOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.models import DAG
from datetime import timedelta
args = {
'owner': 'airflow',
'start_date': airflow.utils.dates.days_ago(2)
}
dag = DAG(
dag_id='example_bash_operator', default_args=args,
schedule_interval='0 0 * * *', # cron style
# 'start_date': datetime.combine(datetime.today() - timedelta(2), time(19, 0, 0)),
dagrun_timeout=timedelta(minutes=60))
cmd = 'ls -l'
run_this_last = DummyOperator(task_id='run_this_last', dag=dag)
run_this = BashOperator(
task_id='run_after_loop', bash_command='echo 1', dag=dag)
#依頼関係1: run_this_lastを実行する前にrun_thisが完了したこと
run_this >> run_this_last
for i in range(3):
i = str(i)
task = BashOperator(
task_id='runme_'+i,
bash_command='echo "{{ task_instance_key_str }}" && sleep 1',
dag=dag)
#依頼関係2: run_thisを実行する前にtask(runme_0, runme_1, rune_2)が完了したこと
task >> run_this
bash_task = BashOperator(
task_id='also_run_this',
bash_command='echo "run_id={{ run_id }} | dag_run={{ dag_run }}"',
dag=dag)
#依頼関係3: run_this_lastを実行する前にbash_taskが完了したこと
bash_task >> run_this_last
if __name__ == "__main__":
dag.cli()
プラグインの書き方と種類
- 基本クラスを継承してクラスを作成
- 種類
- Operator: execute関数に書かれた処理を実行する
- Sensor: poke関数の帰り値がTrueになるまで処理を実行する
- Hook: mysql/bigqueryなど外部システムconnectionのインターフェース
- その他
airflowのコマンド
- パラメーターを渡す
- タスク一覧を正規表現でリストアップ
- テスト
# -tp TASK_PARAMS, TAG01_v1.0: タグ名,
# download_to_bigquery-#{app_id}: タスク名, 2017-08-09: 対象日
airflow test -tp '{"_OVERRIDE_FROM_DATE": "2017-08-01", "_OVERRIDE_TO_DATE": "2017-08-05"}' TAG01_v1.0 download_to_bigquery-#{app_id} 2017-08-09
手動実行
# 初めて過去のデータを処理する（実行順番の注意: 必ず古い順ではない）
airflow backfill -s 2017-05-30T19:00:00 -e 2017-06-31T19:00:00 -I TAG01_v1.0
# 過去のデータを再処理する（実行順番の注意: 必ず古い順ではない）
airflow clear -t 'merge_event.*75019' TAG01_v1.0
注意点
- timezoneはutcなので、日本時間をutcに変更しないといけないです
- 19:00PM UTC yesterday => 04:00 AM JST today
- 昨日の４時に実行したい:datetime.combine(datetime.today() - timedelta(2), time(19, 0, 0))
- start_dateとintervalの混乱
- 最初のタスクが実行されたタイミングはintervalが過ぎた時: start_date + interval
- start_dateかintervalを変更したら、dag_idを変更しましょう
- 再集計時の実行順が選べない
- 一気にタスクがクリアすると、並列実行されるが、でも古い順に実行したい
- 対応：順番に依存しない再集計する別DAGを作成する
apache airflowの利用シーン
- Datawareの事前処理
- レポートティング
- 機械学習(spark,dataflowとの連携)
airflow UI
Tree View
過去タスクの一覧表示される
過去分の一括クリアなどが出来て便利(コマンドの方が便利)
ガントチャート(Gantt)
タスクごとの実行時間が見れて時間のかかっているタスクがわかる
チューニングなどに使う
その他
- kubernetesでリリース
- 失敗したら、slackに通知する
その他
cd kube
make list-pods
kubectl get po -a --namespace airflow-prod
NAME READY STATUS RESTARTS AGE
flower-616933508-5kbxc 1/1 Running 0 17d
postgres-727646253-pxr8b 1/1 Running 0 39d
rabbitmq-1570300244-h251r 1/1 Running 0 39d
scheduler-2020340155-1nplv 1/1 Running 179 16d
web-1384073748-qj44c 1/1 Running 0 16d
worker-1579982757-tvzg3 1/1 Running 0 16d
make list-services
kubectl get svc -a --namespace airflow-prod
NAME CLUSTER-IP EXTERNAL-IP PORT(S) AGE
flower 10.23.XXX.227 <nodes> 5555:32081/TCP 39d
postgres 10.23.XXX.130 <none> 5432/TCP 39d
rabbitmq 10.23.XXX.203 <none> 5672/TCP,15672/TCP 39d
web 10.23.XXX.195 XXX.193.207.XXX 8080:32080/TCP 39d
worker-subdomain None <none> 8793/TCP 39d