Google Cloud ComposerでBigqueryへデータをロードする

## はじめに
業務でデータ基盤の構築をしています。
今回は`Cloud Composer`を用いて、`BigQuery`でcsvデータをロードしてみたいと思います。
使用する諸々は[こちら](https://github.com/HagaSpa/composer)


## TerraformによるComposer環境の構築

`Terraform`を使用して、クラウドリソースを作成すると管理が楽です。
そのため今回は`Terraform`を使用して、リソース作成を行います。

### サービスアカウントの作成
`Terraform`で`Composer`の構築をするために、サービスアカウントからキーを発行しておいてください
[サービスアカウント キーの作成と管理](https://cloud.google.com/iam/docs/creating-managing-service-account-keys?authuser=19&hl=ja)


### Terraformaのインストール
[公式サイト](https://www.terraform.io/downloads.html)からzipを落とせますが、homebrewでインストールすることも出来ます

```
$ brew install terraform
```

### tfファイルの作成

次にComposerを立ち上げるための`.tf`ファイルを作成します。

```composer.tf
# google認証
provider "google-beta" {
  credentials = "${file("account.json")}"
  project     = "project_id"
  region      = "asia-northeast1"
}

# 作成するComposerの内容
resource "google_composer_environment" "composer-environment" {
   name = "sample"
   project = "project_id"
   region = "asia-northeast1"
   config {
      node_count = 3

      node_config {
         zone = "asia-northeast1-c"
         machine_type = "n1-standard-1"
         disk_size_gb = 20
      }
   }
}

```

`provider.credentials`には、発行したキーファイル（.json）のパスを指定します。
ただこの状態だとファイルにセキュアな情報が記載されてしまいます。git管理するときにもめんどそうです。

そのため[Terraform](https://www.terraform.io/docs/providers/google/getting_started.html#adding-credentials)でも勧められているように、以下のように環境変数をセットした方が利便性が高いです。可能ならこちらを実施してください。

```
$ export GOOGLE_CLOUD_KEYFILE_JSON={{path}}
```

ちなみ`provider`で`google-beta`を使用しているのは、python3を`Composer`で使いたかったからです。
特段こだわりなければ`google-beta`ではなく、[google](https://www.terraform.io/docs/providers/google/index.html)で問題ありません。


### リソースの作成
上記で作成した`.tf`ファイルがあるディレクトリに移動して、`init` > `apply`を行います。

```
$ terraform init
$ terraform apply
...
Do you want to perform these actions?
  Terraform will perform the actions described above.
  Only 'yes' will be accepted to approve.

  Enter a value:
```

`yes`　でリソース作成を行います。

```
  Enter a value: yes
google_composer_environment.composer-environment: Still creating... (10s elapsed)
google_composer_environment.composer-environment: Still creating... (20s elapsed)
google_composer_environment.composer-environment: Still creating... (30s elapsed)
.
.
.
Apply complete! Resources: 1 added, 0 changed, 0 destroyed.
```

`Composer`が作成されるまで10分ほど掛かります。気長に待ちましょう！
作成されるとGCPのコンソールから「Composer」を選択すると、下記のように環境が作成されているのを確認できます。

![スクリーンショット 2019-01-19 13.56.49.png](https://qiita-image-store.s3.amazonaws.com/0/128338/efa898ea-0cbe-4b50-d1df-99895fd37794.png)




## Dagファイルの作成
`Composer　（Airflow）`では、ワークフローの定義をDagファイルというものに記載します。ただのpythonファイルです。
今回は`gcs`にある`csv`ファイルを`BigQuery`に投入するだけのタスクを持つ、ワークフローを作成します。

やりたいことは`BigQuery`にデータを投入することなので、`Dataflow`を`Composer`から呼び出すような実装にします。
`Dataflow`はGCPで用意されている、フルマネージドなデータ処理サービスです。

`Dataflow`にはgoogle側でいくつかテンプレートが用意されており、今回は`gcs`にあるcsvをBigQueryへロードするため、`Text Files on Cloud Storage to BigQuery`を使います。

<img width="1679" alt="スクリーンショット 2019-01-21 19.52.07.png" src="https://qiita-image-store.s3.amazonaws.com/0/128338/02137a4d-ddce-5369-7c75-e8e8b5d3f7df.png">

上記のテンプレートを`Composer`から呼び出して、データのインポート作業は`Dataflow`に任せます。
Dagファイルは以下のようになります。

```csv_to_bigquery.py

from airflow import DAG
from airflow.contrib.operators.dataflow_operator import DataflowTemplateOperator
from datetime import datetime, timedelta
from airflow import models

# gcpのプロジェクトID
PROJECT_ID = models.Variable.get('project_id')
# gcsのパス
BUCKET_NAME = models.Variable.get('gcs_bucket_name')
# dag名
DAG_NAME = 'csv_to_bigquery'

yesterday = datetime.combine(
    datetime.today() - timedelta(days=1) + timedelta(hours=9),
    datetime.min.time()
)

default_dag_args = {
    'owner': 'HagaSpa',
    'start_date': yesterday,
    'project_id': PROJECT_ID,
    'dataflow_default_options': {
        'project': PROJECT_ID,
    }
}

with models.DAG(
    dag_id=DAG_NAME,
    schedule_interval="@once",
    default_args=default_dag_args) as dag:
    
        t1 = DataflowTemplateOperator(
            task_id='task1',
            template='gs://dataflow-templates/latest/GCS_Text_to_BigQuery',
            parameters={
                # udf.jsファイルにある呼び出したいメソッド名
                'javascriptTextTransformFunctionName': 'transform',
                # bqのスキーマ定義ファイルのgcsパス               
                'JSONPath': 'gs://{}/composer/schema/schema.json'.format(BUCKET_NAME),
                # udf.jsファイルのgcsパス
                'javascriptTextTransformGcsPath': 'gs://{}/composer/udf/udf.js'.format(BUCKET_NAME),
                # bqへ投入するcsvファイルのgcsパス
                'inputFilePattern': 'gs://{}/composer/csv/sample.csv'.format(BUCKET_NAME),
                # 保存するbqのプロジェクトid:データセット名.テーブル名        
                'outputTable': '{}:my_dataset.sample'.format(PROJECT_ID),                              
                # bqへのロード中のtempディレクトリ                
                'bigQueryLoadingTemporaryDirectory': 'gs://{}/composer/temp'.format(BUCKET_NAME),      
            },
        )

```

- まず`PROJECT_ID = models.Variable.get('project_id')`ですが、`model.Variables`は`Airflow`で定義できる環境変数のことです。プロジェクトIDや`gcs`のバケット名など、外だししておく方が運用が楽な値は環境変数に定義します。
各値は自身の環境に合わせて、`Airflow`の環境変数をセットして下さい。
「環境変数はどうやって定義すんのや？」って方は[こちら](https://airflow.apache.org/ui.html#variable-view)を参照ください。


- `with models.DAG()`がairflowにおける、１つのDagになります。
SubDag？とか使えば、複数記載が出来るようですが今回はこれだけになります。


- `DataflowTemplateOperator()`が、Dataflowのテンプレートを呼び出している箇所になります。
`parameters`に記載している内容は、Dataflowのテンプレート（GCS_Text_to_BigQuery）を使用する上で必須なものです。
[Dataflow](https://cloud.google.com/dataflow/docs/guides/templates/provided-templates?hl=ja#cloud-storage-text-to-bigquery)によるものです。

- `udf.js`は、「ユーザが定義した関数ファイル」です。具体的には「ファイルの行を引数とした、jsファイル」となっており、BigQueryへcsvファイルを取り込む際のカラム名とフィールド値をマッピングします。


```udf.js
function transform(line) {
    var values = line.split(',');
    
    var obj = new Object();
    obj.id = values[0];
    obj.name = values[1];
    obj.age = values[2];
    var jsonString = JSON.stringify(obj);
    
    return jsonString;
}

```

```sample.csv
// id,name,age
1,hagaspa,26
2,anyone,99
3,hoge,17

```

上記2ファイルを`Dag`ファイルに記載された、`gcs`のパスに配置してください。
その後`Composer`のコンソールから「Dag　のフォルダ」をクリックし、表示された`gcs`にDagファイルをアップロードしてください。


### Dagの起動
Dagファイルをアップロードしてから、しばらく経つと`Airflow`の画面に以下のようにDagが表示されます。
（おそらく初回は自動的に開始されているはずです。開始されていない場合は、右側のRunをクリックして下さい。）
<img width="918" alt="スクリーンショット 2019-01-21 20.18.34.png" src="https://qiita-image-store.s3.amazonaws.com/0/128338/a0851519-44a0-fbea-fb74-887abfa66d59.png">

Recent Tasksがグリーンになると成功です。
`Airflow`が`Dataflow`を起動し、BigQueryへデータをインポートしているか確認しましょう！
まずはDataflowですが、成功していると下記のようになっているはずです。

<img width="1228" alt="スクリーンショット 2019-01-21 20.27.38.png" src="https://qiita-image-store.s3.amazonaws.com/0/128338/97c0a7f5-0610-7f2b-bd44-d622159e740c.png">

基本的にオールグリーンになっていれば成功ですが、ありがちなミスとして`gcs`のファイル名を間違えているとエラーにならずに0KBを読み取るということをするので、`Read from source`のフェーズは要チェックしましょう。

最後に`BigQuery`にデータをインポート出来ているか確認します。

<img width="881" alt="スクリーンショット 2019-01-21 20.19.52.png" src="https://qiita-image-store.s3.amazonaws.com/0/128338/70b5fe5a-f7ea-7e64-6be0-3da0b8beeb43.png">

csvデータがインポート出来ました！

## まとめ
データ基盤を作る上では、しばしばテーブルの更新状態に依存することがあります。
今回は単純にデータをロードしただけですが実業務だと、

1. 前日データを使って*テーブルA*にデータをロードする。
2. *テーブルA*が更新されているなら、`Tableau`データソースへ適用や集計処理などを実施。
3. *テーブルA*への取り込みがエラーなら、後続処理を止めSlack通知を行う。

などのワークフローを組むことが多いです。
そのため`Airflow`もとい、`Composer`を使用してデータ基盤構築フローを組むことはとても便利だと感じました。


## 困ったら
基本は[ドキュメント](https://airflow.apache.org/)に記載されています。またossなので最悪[ソースコード](https://github.com/apache/airflow)を読めます。
`Dataflow`以外にも色んなサービスを`Airflow`で使用できるので、[operators](https://github.com/apache/airflow/blob/master/airflow/contrib/operators/s3_to_gcs_operator.py)を参照してやりたいことが`Airflow`で実現できるか考慮しましょう！