Cloud Composerを使ってデータ分析基盤を作ったときの話

本記事は、[VISITS Advent Calendar 2019](https://qiita.com/advent-calendar/2019/visits)の10日目の記事です。
昨日は、@muraokaz　による[クラスタリング＋二次元マッピングによる可視化を定量評価する](https://qiita.com/muraokaz)　でした。

VISITS Technologiesでインフラエンジニアをしている@syogunです。
本日は、[Cloud Composer](https://cloud.google.com/composer/?hl=ja)を使用してVISITSでデータ分析基盤を構築(in progress)した時の話をさせて頂ければと思います。
主に`CloudSQL(MySQL)→GCS→BigQuery`の部分について書きます。
人様の情報がベースで恐縮ですが、[こちらの記事]((https://medium.com/grensesnittet/copy-data-from-cloud-sql-to-bigquery-using-apache-airflow-cloud-composer-part-2-33aa02bf456a))を参考にして構築を進めました。
Cloud ComposerがGKE上で動いているため、ちょっとGKEの話も出てきます。
では参りましょう。

# 1. 構成図
今後変更はあるかもしれませんが、想定している構成図はこんな感じです。（GAEのログ転送は含まない）
<img width="1484" alt="スクリーンショット 2019-12-02 18.08.11.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/95324/80c18cf4-840b-3ee9-faca-434d6de69405.png">

データマスキングの部分は、GCPの[DLP](https://cloud.google.com/dlp/?hl=ja)を使用してメールアドレスや電話番号を判別して、
その情報を元にデータをマスキングする方法を検討中です。
現在進行形なので、仕組みができたらまた書こうと思います。

# 2. 背景
ユーザーや弊社サービス利用企業のActivity等を分析し、プロダクトの改善に活かしていきたいというニーズがあり、データ分析基盤を作ろう！という話になりました。
具体的には、アプリケーションログ(GAE)、Google Analytics、DB(MySQL)をDataSourceとし、
そのDataSourceに必要なマスキング処理を行い、DWHにロードするといった構成を考えました。
弊社のクラウドは主にGCPを使っているため、DWHは文句なしでBigQueryを使うことにしました。

# 3. いろいろ試しました
そもそも「何を使えばいいだ？」というレベルから始まったのいろいろ試行錯誤しました。
おそらく一番の課題は、データを抽出、加工するところだと考え、
最初は、[Cloud DataPrep](https://cloud.google.com/dataprep/?hl=ja)や[Cloud DataFusion](https://cloud.google.com/data-fusion/?hl=ja)を使用して、ノンプログラミングで楽することを考えました。
しかしその淡い夢は、以下の理由で絶たれれました。。

## Cloud DataPrepについて
DataPrepとは、公式ドキュメントを見ると

> ブラウザ環境の簡単なドラッグ＆ドロップ操作で、分析用の多様なデータセットを視覚的に探索、クリーニング、準備できるインテリジェントなデータ準備およびクレンジング サービスです。

とあります。
DataSourceのマスキングにこのDataPrepを使うことを検討したんですが、

* そもそも対応しているDataSourceがGCSとBigQueryしかなく、Cloud SQLに対応していなかった。
* Cloud SQLからのデータ抽出は、別途CircleCI等のCIツールから`gcloud sql export csv`を叩いてGCSにCloud SQLのデータを出力することも検討したが、負けた感が強くて途中でやめた。

という理由で無念の撤退。

## Cloud DataFusionについて
DataFusionとは、公式ドキュメントを見ると

> ETL および ELT のデータ パイプラインを効率的に構築して管理できる、フルマネージドかつクラウド ネイティブなデータ統合サービスです。

とあります。
次にDataPrepと比較して様々なDataSourceに対応しているDataFusionを検討したんですが、

* DataFusionからCloud SQLへの接続がどうしてもできなかった。。涙
    * [こちら記事]((https://medium.com/google-cloud-jp/cloud-data-fusion%E3%81%A7aws%E3%81%AEmysql%E3%81%8B%E3%82%89bigquery%E3%81%AB%E3%81%A4%E3%81%AA%E3%81%92%E3%81%A6%E3%81%BF%E3%81%9F-4dd02782f278))だとRDSはできそうだった..

という理由で無念の撤退。
余談ですが、DataFusionは、DataSourceとしてS3とかにも対応しており、
S3->BigQueryが実現できそうでした。これはAWSユーザにとっては嬉しいですね。

# 4. 最終的に何を使ったのか？
そんなこんなで最終的にCloud Composerにたどり着きました。
理由としては、

* ワークフローを定義できるのでマスキング等の処理を任意のタイミングで挟める
* `MySqlToGoogleCloudStorageOperator`や`GoogleCloudStorageToBigQueryOperator`といったOperator(特定のプログラムを実行するためのテンプレートのようなもの)が用意されている
* すでにCloudSQL(MySQL)→GCS→BigQueryをやっている人がおり、[サンプルのDAGファイル](https://github.com/ael-computas/gcp_cloudsql_airflow_bigquery/blob/master/dags/copy_cloudsql_to_bigquery_v2.py)もあった。
    * Airflowは、個人的になかなか学習コストが高かったので助かった

が挙げられます。

# 5. Cloud Composerセットアップな大まかな流れ
以下のような流れでセットアップできます。

1. Cloud Composer環境を構築
2. Cloud SQL ProxyをGKE上にデプロイ
3. Airflow connectionsの設定
4. DAGファイルの作成、設置

# 6. Cloud Composerで`CloudSQL(MySQL)→GCS→BigQuery`を実現するためのポイント
ここでは、私がハマった箇所をご紹介します。

## 1. Cloud Composer環境を構築
当初は、以下のgcloudコマンドでCloud Composer環境を構築しました。

```
$ gcloud composer environments create <your env name> --location asia-northeast1 
```

しかしながら、`Composer Backend timed out.` なるエラーが出てコマンドがこけます。
原因は、Cloud Composer環境作成の際、サービスアカウントを明示的に指定しないと
デフォルトでは、`<your project number>-compute@developer.gserviceaccount.com` というサービスアカウントが使用されます。
このサービスアカウントに`Composer ワーカー`権限がないため、Composerの作成にこけるようです。（エラーメッセージはなんとかしてほしい）

デフォルトのサービスアカウントは使いたくなかったので、
今回は別途`cloud-composer@<your project name>.iam.gserviceaccount.com`を作成し、そちらでComposerを作成しました。以下、コマンド例。

```
$ gcloud composer environments create <your env name> --service-account=cloud-composer@<your project name>.iam.gserviceaccount.com --location asia-northeast1 
```

サービスアカウント権限は、

* `Composer ワーカー`
* `BigQuery 管理者`
* `クラウドSQL編集者`
* `ストレージ オブジェクト管理者`

をつけました。

## 2. Cloud SQL ProxyをGKE上にデプロイ
Cloud ComposerからCloud SQLへの接続は、[Cloud SQL Proxy](https://cloud.google.com/sql/docs/mysql/sql-proxy?hl=ja)を使用します。
Cloud Composer環境を作成すると、裏でGCEが起動し、その上でKubernetesが起動することになります。
よって、Cloud SQL Proxy用の`Deployment`,`Service`を作成し、DBの接続設定をCloud SQL ProxyのEndpointに向けることでCloud SQLへの接続が可能となります。

まず、kubectlコマンドでサービウアカウントのsecretを作成します。

```
$ mv <your service account file> credentials.json
$ kubectl create secret generic <your secret name> --from-file=key.json=credentials.json
```

次に[こちらのManifest File](https://github.com/ael-computas/gcp_cloudsql_airflow_bigquery/blob/master/yaml/sql_proxy_mysql_deployment.yaml)を参考にCloud SQL Proxy用のDeploymentを作成します。
こちらのManifest Fileに少し手を加えて、さきほど作成したsecret(サービスアカウントキー)を埋め込んでます。
参考までに手を加えた部分を抜粋します。

```yaml

    spec:
      volumes:
        - name: ssl-certs
          hostPath:
            path: /etc/ssl/certs
        - name: google-cloud-key
          secret:
            secretName:  <your secret name> # secret名を指定
      containers:
        - image: gcr.io/cloudsql-docker/gce-proxy:1.11
          volumeMounts:
            - name: ssl-certs
              mountPath: /etc/ssl/certs
            - name: google-cloud-key # secretをmount
              mountPath: /var/secrets/google
          env:
            - name: GOOGLE_APPLICATION_CREDENTIALS # GOOGLE_APPLICATION_CREDENTIALSにサービスアカウントを設定（不要かも？）
              value: /var/secrets/google/key.json
          command:
            [
              "/cloud_sql_proxy",
              "-instances=visits-for-innovators-prd:asia-northeast1:innovators-production-replica=tcp:0.0.0.0:3306",
              "-credential_file=/var/secrets/google/key.json", # credentialを明示的に指定
            ]

```

次に、[こちらのManifest File](https://github.com/ael-computas/gcp_cloudsql_airflow_bigquery/blob/master/yaml/sql_proxy_dvh_service.yaml)を`kubectl apply`してServiceリソースを作成します。
ここで作成したServiceがCloud SQL ProxyのEndpointになります。

## 3. Airflow connectionsの設定
GCPのCloud ComposerからAirflowのGUIに接続できます。
このAirflowのGUIからCloud SQLのConnections設定を行います。
ここでは文字コードを指定しないとExportしたjsonが**文字化け**することがあったんですが、以下のように`utf8`を明示的に指定することで回避できました。
![スクリーンショット 2019-12-03 11.55.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/95324/7cbca8ba-cf84-9e88-9f98-6d4bf0bd0948.png)

なお、`Conn ID`には、上記で作成した`Serviceリソース名.default`(例 mysql-dvh-sqlproxy-service.default)を指定すればOKです。
あとは、KubernetsのKube-DNSが名前解決してくれるはずです。


## 4. DAGファイルの作成、設置
Cloud Composerを作成すると自動でDAGファイルを配置するGCSバケットが作成されます。
このGCSバケットにDAGファイルを配置すれば、ワークフローが読み込まれます。
ここでのハマりポイントは、DAGファイルでSchedule設定しているのに、当該時刻が過ぎてもワークフローが走らないことがありました。
AirflowのSchedule設定のポイントは以下2つでした。

#### スケジュール実行
##### start_dateを設定する
まず、`start_date`、ワークフローの開始日付を設定しないと動きません。

```python

# 設定例
default_args = {
    'owner': '<your name>',
    'depends_on_past': False,
    # Change this to a literal date to make scheduler work.
    'start_date': datetime(2019, 11, 25),　# ここをちゃんと設定する
    'email': ['<your email address>'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=1),
}
```

##### スケジュール実行時間の計算方式を理解する
Scheduleを設定してもジョブが動かないず、少しハマりました。
詳細は、[こちらの記事](https://medium.com/programming-soda/apache-airflow%E3%81%A7%E3%82%A8%E3%83%B3%E3%83%89%E3%83%A6%E3%83%BC%E3%82%B6%E3%83%BC%E3%81%AE%E3%81%9F%E3%82%81%E3%81%AE%E6%A9%9F%E6%A2%B0%E5%AD%A6%E7%BF%92%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%82%92%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B-part4-590ad9f9fb80)の解説がすごく分かりやすかったですが、ポイントは￥


> start_date + schedule_interval が経過したらスケジュールされる
> スケジュールされるタイミングはschedule_intervalの経過後

とのことでした。
このあたり把握せず、デバッグしようと思うとハマりますのでお気をつけを。

#### BigQueryOperatorを使ったデータマートの作成
BigQuery上のTableをjoinして、データマート的なものを作ろうと考えました。
これには、`AirflowのBigQueryOperator`を使用することを考えています。
具体的には、既存のBigqueryのTableに対し`sql='sql/template.sql',`で実行したSQLの結果を、`destination_dataset_table`で指定したDatasetのTableに出力するというものです。
ところが、上記のQuery実行結果が、`destination_dataset_table`で指定したDatasetのTableにコピーされない事象がありました。
これは、`create_disposition='CREATE_IF_NEEDED',` というパラメータをつけることで解決しました。
以下サンプルDAGです。
なお、ここでしているSQLは、DAGファイルと同じGCS上のディレクトリに配置すればOKです。

```python

def gen_transform_table_task(table_config):
    transform_task = BigQueryOperator(
        task_id='transform_of_{}'.format(table_config.params['export_table']),
        sql='sql/template.sql',
        destination_dataset_table="{}.{}.{}".format(table_config.params['gcp_project'],
                                                    table_config.params['stage_dataset'] +
                                                    "_summary",
                                                    table_config.params['stage_table'] + "_{{ ds_nodash }}"),
        params={"project_nm": "<your gcp project name>",
                "dataset_nm": "innovators",
                "table_nm": "{}".format(table_config.params['export_table'])},
        write_disposition='WRITE_TRUNCATE',
        create_disposition='CREATE_IF_NEEDED', # ここ
        bigquery_conn_id='<your bigquery conn id>',
        location='US',
        use_legacy_sql=False,
        dag=dag)

    transform_task.doc_md = """\
        # Import table from storage to bigquery
        task documentation
        """
    return transform_task
```


# 7. まとめ
Cloud Composerは、いろんなOperatorが用意してあるので、使い方をマスターできればかなり便利そうな感じですね。
これからいろいろキャッチアップしていきたいです。

明日は、@ken_hikita による「kintoneの承認経路を作ってみたときのお話し」です！

# 8. 参考情報
以下の文献を参考にさせていただきました。

* [Cloud Data FusionでAWSのMySQLからBigQueryにつなげてみた](https://medium.com/google-cloud-jp/cloud-data-fusion%E3%81%A7aws%E3%81%AEmysql%E3%81%8B%E3%82%89bigquery%E3%81%AB%E3%81%A4%E3%81%AA%E3%81%92%E3%81%A6%E3%81%BF%E3%81%9F-4dd02782f278)
* [Copy data from cloud SQL to BigQuery using apache airflow/cloud composer part 2](https://medium.com/grensesnittet/copy-data-from-cloud-sql-to-bigquery-using-apache-airflow-cloud-composer-part-2-33aa02bf456a)
* [Apache Airflowでエンドユーザーのための機械学習パイプラインを構築する Part4](https://medium.com/programming-soda/apache-airflow%E3%81%A7%E3%82%A8%E3%83%B3%E3%83%89%E3%83%A6%E3%83%BC%E3%82%B6%E3%83%BC%E3%81%AE%E3%81%9F%E3%82%81%E3%81%AE%E6%A9%9F%E6%A2%B0%E5%AD%A6%E7%BF%92%E3%83%91%E3%82%A4%E3%83%97%E3%83%A9%E3%82%A4%E3%83%B3%E3%82%92%E6%A7%8B%E7%AF%89%E3%81%99%E3%82%8B-part4-590ad9f9fb80)