Cloud SchedulerでCloud Dataflowのジョブを定期的に起動する



* [DataflowのテンプレートはAPIでジョブを起動出来ます](https://cloud.google.com/dataflow/docs/guides/templates/running-templates?hl=ja#rest-api-)
* [Cloud ScedulerはHTTP(s)エンドポイントを呼べます](https://cloud.google.com/scheduler/docs/?hl=ja)

ということは、Cloud SchedulerからDataflowのジョブをキック出来る？と思い、試してみました。


# 準備

* Dataflow APIと、テンプレートが使う[適当なAPI(Pub/Subとか)を有効化](https://cloud.google.com/dataflow/docs/quickstarts/quickstart-templates?hl=ja#before-you-begin)
* Cloud SchedulerがDataflow APIを叩くためのサービスアカウントを作成する
 * [Dataflow管理者](https://cloud.google.com/dataflow/docs/concepts/access-control)に類する権限が必要
* 入力・出力のバケット作る

# 設定

[projects.locations.templates.create](https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.locations.templates/create)を叩き、[Bulk Compress Cloud Storage Files（Google提供のテンプレート）](https://cloud.google.com/dataflow/docs/guides/templates/provided-utilities#bulk-compress-cloud-storage-files)をキックする設定をします。

下例では、[gcloudコマンド](https://cloud.google.com/sdk/gcloud/reference/beta/scheduler/jobs/create/http?hl=ja)で設定していますが、GCPコンソールからでも出来ます。

```
gcloud beta scheduler jobs create http job_name \
--schedule='every day 09:00' \
--uri="https://dataflow.googleapis.com/v1b3/projects/プロジェクトID/locations/asia-northeast1/templates" \
--message-body-from-file="dataflow_message_body.json" \
--oauth-service-account-email='サービスアカウントの名前'
```

最低限、設定するのは

* スケジュール間隔
* Cloud Schedulerが叩くエンドポイント
 * https://dataflow.googleapis.com/v1b3/projects/プロジェクトID/locations/asia-northeast1/templates
* ジョブをキックするサービスアカウント
* HTTP Bodyを指定するファイル
 * （--message-bodyにするとコマンドライン引数でも指定できます）

です。


HTTP Bodyに、[Dataflow API一般の設定（gcsPathとやjobName）](https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.locations.templates/create)と、各テンプレート固有のパラメータ（parameters）を設定します。

```
cat dataflow_message_body.json | jq                                                                                                                         1739ms  月  1/13 15:32:41 2020
{
  "jobName": "dataflow_job_name",
  "parameters": {
    "inputFilePattern": "gs://バケット名/dataflow/compress_input.txt",
    "outputDirectory": "gs://バケット名/dataflow/compress_output.txt.gz",
    "outputFailureFile": "gs://バケット名/dataflow/failure",
    "compression": "GZIP"
  },
  "gcsPath": "gs://dataflow-templates/latest/Bulk_Compress_GCS_Files"
}
```

# 実行
スケジュールされた時間になるまで待つか、待ち遠しい人はCloud Scheduler画面から「今すぐ実行」しましょう。

# 確認
ジョブが出来たか、見てみます。


```
gcloud beta dataflow jobs list                                                                776ms  月  1/13 15:52:22 2020
`--region` not set; getting jobs from all available regions. Some jobs may be missing in the event of an outage. https://cloud.google.com/dataflow/docs/concepts/regional-endpoints
JOB_ID                                    NAME                TYPE   CREATION_TIME        STATE      REGION
2020-01-12_22_32_14-12998664790411440361  dataflow_job_name   Batch  2020-01-13 06:32:16  Done       asia-northeast1
```

ジョブが出来ています（GCPコンソールでジョブ実行状況を見られます）。

結果のファイルも見てみましょう。

```
gsutil cat gs://qiita_test/dataflow/compress_output.txt.gz/compress_input.txt.gz | file -                                                                   1876ms  月  1/13 15:35:45 2020
/dev/stdin: gzip compressed data, from FAT filesystem (MS-DOS, OS/2, NT)

gsutil cat gs://qiita_test/dataflow/compress_output.txt.gz/compress_input.txt.gz | gzcat                                                                    2029ms  月  1/13 15:35:52 2020
テスト入力
```
圧縮出来ていますね。

# 後始末
ジョブが作られ続けるので、動作確認終わったらCloud Schedulerのジョブは削除しましょう。


```
gcloud beta scheduler jobs delete job_name                                                      
```


# 補足
「projects.locations.templates.create」と「projects.templates.create 」という２つのAPIがあります。

template.createのドキュメントでは説明を見つけられなかったのですが、[projects.jobs.create](https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.jobs/create)の方には、

> To create a job, we recommend using projects.locations.jobs.create with a regional endpoint. Using projects.jobs.create is not recommended, as your job will always start in us-central1.

とあるので、「projects.locations.templates.create」の方を使った方がいいのかなーと思っています。





# Dataflowジョブの停止
Dataflowジョブの停止（cancel/drain）も[APIがある](https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.locations.jobs/update)ので（※）、CloudSchedulerからジョブを停止することも出来ます。
ただし、自動で割り振られる**job idの指定が必要**で、定期的にジョブを起動し、定期的にそれを止めるような処理は出来ないので、実用性は低い気がします…


※ リクエストbodyで[requestedStateをdrainingに指定します](https://stackoverflow.com/questions/48413228/how-do-you-drain-a-dataflow-job-through-the-gcloud-api/50819642)