Cloud Composer を使ったワークフロー開発に便利なコマンドまとめ #GoogleCloud

Web UI からポチポチしていると日が暮れるのでコマンドラインから開発できるように調べたまとめ

デプロイ

gcloud composer environments storage {dags,plugins} import を使ってファイルをアップロードする。うちではディレクトリ構造を合わせて作って以下のようにデプロイしている。

不便なことに {dags,plugins} ディレクトリを丸ごとアップロードすることができず、--source でサブディレクトリを指定しないといけないため、find | xargs でサブディレクトリ一覧を作ってから import している。

find "composer/dags" -maxdepth 1 -mindepth 1 | xargs -t -P4 -I{} \
  gcloud composer environments storage dags import \
  --environment ${COMPOSER_ENVIRONMENT_NAME} \
  --location ${LOCATION} \
  --source {}

find "composer/plugins" -maxdepth 1 -mindepth 1 | xargs -t -P4 -I{} \
  gcloud composer environments storage plugins import \
  --environment ${COMPOSER_ENVIRONMENT_NAME} \
  --location ${LOCATION} \
  --source {}

ゴミDAGもしくはゴミファイルの掃除

不便なことに import は上書きしかしてくれないため、ゴミファイルの掃除は別途しなければならない。

また、不要になったDAGを削除する際は、Composer 用の GCS からファイルを削除した上で、Airflow の delete_dag コマンドを使って、Airflow の MySQL からエントリを削除しなければならない。

ゴミDAGの削除

gcloud composer environments storage dags delete "${DAG}.py" \
  --environment ${COMPOSER_ENVIRONMENT_NAME} \
  --location ${LOCATION}

sleep 15

gcloud composer environments run \
  ${COMPOSER_ENVIRONMENT_NAME} --location ${LOCATION} \
  delete_dag -- "${DAG}"

ゴミ plugins ファイルの削除

gcloud composer environments storage plugins delete "${FILE}" \
  --environment ${COMPOSER_ENVIRONMENT_NAME} \
  --location ${LOCATION}

また storage {dags,plugins} list コマンドでファイルの一覧を参照できます。

TODO: sync してくれるようなスクリプトの作成

DAGの特定タスクの実行

airflow の test コマンドを使って特定タスクの実行ができる。これを使って開発するかしないかで開発効率が全く違う。

gcloud composer environments run \
  ${COMPOSER_ENVIRONMENT_NAME} --location ${LOCATION} \
  test -- my_dag my_task 2019-08-23T00:00:00+0000

DAG全体の実行

airflow の trigger_dag コマンドを使ってDAG全体に実行ができる。execution_time にユニーク制約があるので開発中は秒を 01 にしたり 02 にしたりすることがよくある。

gcloud composer environments run \
  ${COMPOSER_ENVIRONMENT_NAME} --location ${LOCATION} \
  trigger_dag -- my_dag -e 2019-08-23T00:00:00+0000

Airflow WebUI URLの取得

Composer を作り直すと Airflow WebUI の URL が変わってしまうので次のように取得すると便利。

gcloud composer environments describe \
  ${COMPOSER_ENVIRONMENT_NAME} --location ${LOCATION} \
  | grep airflowUri | awk -F': ' '{print $2}'

macOS なら open <URL> コマンドでブラウザを開ける

DAGファイルの Syntax チェック

See Cloud Composer 向けの DAG ファイルの Syntax チェックをローカルで実行する