Cloud Dataflowを用いてデータの値によって動的に宛先を変えてGCSに保存する

# はじめに
この記事は [Classi Advent Calendar 2019](https://qiita.com/advent-calendar/2019/classi)の23日目の記事です。

こんにちは、ClassiのデータAI部の@tomoyanamekawaです。
普段はGCP上でのデータ分析基盤構築をメインでやっています。

最近、「BigQueryにあるデータを中の値に合わせてファイル分割してGCSに保存したい」ということがあって、その時にCloud Dataflowにお世話になりました。
他の方にも需要ありそうで、かつPythonでの実装例が少なかったのでまとめようと思います。

# 今回のゴール
BigQueryにある特定のテーブルをGoogle Cloud Storage(GCS)にexportする処理をdailyで実行する。
ただしあるcolumnの値によって保存先のディレクトリを変えたい。
ファイル形式はjson。

## 例
BigQueryにあるreservationsテーブルを
<img width="400" alt="reservationsテーブル" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/256675/4163bef3-2780-9fe2-f7ca-76447b494f98.png">
このように日付/shop_idごとに分けてGCSに保存をしたい。
<img width="400" alt="reservations_GCS" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/256675/3ba3762a-9b47-867a-9e2a-6f0e17ed622c.png">


## 完成図

<img width="400" alt="完成図" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/256675/f1461996-ccda-fea3-f94c-8d8159050f91.png">

## 環境
- Python 3.7.3
- apache-beam==2.16.0



# Cloud Dataflowとは
GCPで提供されているサーバレスにETL処理を行えるサービスです。
裏側では[Apache Beam](https://beam.apache.org/)が動いているので、サーバレスにApache Beamを使えるサービスとも言えます。
並列処理を行えるため大規模なデータに対しても高速に処理を行うことができます。

ストリーム処理とバッチ処理の両方に対応していますが、今回はバッチ処理を使います。
詳しくは[公式ページ](https://cloud.google.com/dataflow/?hl=ja)で。

とりあえず使えるようになりたいという方はゆずたそさんの発表資料にある[こちらの手順](https://speakerdeck.com/yuzutas0/20190719?slide=49)が良いと思います（私もこれでキャッチアップしました）。

# カスタムテンプレートの作成
Cloud Dataflowでは「テンプレート」と呼ばれるものを利用してETL処理を作成します。
一般的な処理であれば[Googleが提供するテンプレート](https://cloud.google.com/dataflow/docs/guides/templates/provided-templates?hl=ja)を使用すればGUIベースで簡単にできます。
ただ今回やりたいことはこちらだとできないのでカスタムテンプレートを自分で作成します。

ちなみにプログラミング言語はJavaかPythonが利用できます。
今回はPythonで書きますが、Javaのほうが機能やドキュメントが豊富なので、自分やチームメンバーがJavaを書けてメンテナンスの問題もない場合はJavaのほうがいいと思います。

カスタムテンプレートの中身はこちらです。

```python:test_template.py
import os
import json
import datetime

import apache_beam as beam
from apache_beam.io import fileio
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions, GoogleCloudOptions

        
class JsonSink(fileio.TextSink):
    def write(self, record):
      self._fh.write(json.dumps(record).encode('utf8'))
      self._fh.write('\n'.encode('utf8'))


if __name__ == '__main__':
    now = datetime.datetime.now().strftime('%Y%m%d')
    project_id = 'your_project'
    dataset_name = 'your_dataset'
    table_name = 'your_table'
    bucket_name = 'your_bucket'

    # オプション
    pipeline_options = PipelineOptions()
    google_cloud_options = pipeline_options.view_as(GoogleCloudOptions)
    google_cloud_options.project = project_id
    google_cloud_options.job_name = 'myjob'
    google_cloud_options.staging_location = f'gs://{bucket_name}/staging'
    google_cloud_options.temp_location = f'gs://{bucket_name}/temp'
    google_cloud_options.template_location = f'gs://{bucket_name}/templates/test_template'
    pipeline_options.view_as(StandardOptions).runner = 'DataflowRunner'
        
    #パイプラインの作成
    pipeline = beam.Pipeline(options=pipeline_options)
    (pipeline 
        | 'read' >> beam.io.Read(beam.io.BigQuerySource(
            project=project_id, 
            use_standard_sql=True, 
            query=f'select * from `{project_id}.{dataset_name}.{table_name}`'
        ))
        | 'write' >> beam.io.fileio.WriteToFiles(
            path=f'gs://{bucket_name}/{now}',
            destination=lambda record, table_name=table_name: f"shop_id_{record['shop_id']}/",
            sink=JsonSink(),
            file_naming=beam.io.fileio.destination_prefix_naming()
        )
    )

    pipeline.run()
```

ポイントはこちらの[Dynamic Destinations](https://beam.apache.org/releases/pydoc/2.16.0/apache_beam.io.fileio.html#dynamic-destinations)という機能を利用しているところです。
recordという変数の中に各レコードごとの値が入ってくるので`record['shop_id']`でレコードごとにdestination（保存先のファイル名）を変えられます。

作成したテンプレートはGCS上に置く必要があるのでこちらのコマンドを実行します。

```bash
python -m test_template
```

そうすると`google_cloud_options.template_location`で指定した場所にテンプレートが設置されます。
[実行時にテンプレートの置き場所を設定することもできます](https://cloud.google.com/dataflow/docs/guides/templates/creating-templates?hl=ja#header_3)。

# dailyで動くようにする
Cloud Dataflow自体にはscheduler機能がないため、dailyで実行するためには外部から実行する必要があります。
そこで今回はCloud Scheduler + Cloud Pub/Sub + Cloud Functionsでサーバレスに実行できるようにします。
<img width="400" alt="schedulerの構成" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/256675/1d4af4d9-7dc4-c145-adae-d242c51491e8.png">

Cloud Functionsに下記のスクリプトを登録します。
このスクリプトがカスタムテンプレートを実行してくれます。

```python
from googleapiclient.discovery import build
def main(data, context):
    job = 'my_job'
    dataflow = build('dataflow', 'v1b3')
    request = dataflow.projects().templates().launch(
        projectId='your_project',
        gcsPath='gs://your_bucket/templates/test_template'
    )
    response = request.execute()
```
Cloud FunctionsのトリガーはPub/Subです。
また、Pub/Subをトリガーにする場合には2つ引数を受け取る必要があるので`main(data, context)`としています。

あとはトリガーにしてあるPub/SubのTopicを作成して、Cloud SchedulerからdailyでそのTopicがpublishされるようにすれば完成です！

もしCloud Composerやサーバを立ててその他のワークフローエンジンやcronでScheduleする場合は下記のgcloudコマンドからカスタムテンプレートを実行できます。

```bash
gcloud dataflow jobs run my_job \
        --gcs-location gs://your_bucket/templates/test_template \
        --region=asia-northeast1
```

# おわりに
このような処理を大規模を短時間で行えるシステムを自分たちで実装しようと思うと手間が恐ろしいのでCloud Dataflowはとても便利です。
ただちょいとお高いので、「Cloud Dataflowでxx万円とかした」とかならないように使い所は選ぶ必要があるかなと思っています。

あしたは@tetsuya0617さんです。お楽しみに！