GCEでdigdag+embulkのETL基盤を構築してみた

# この記事はなに？

データ基盤を構築する上で、必ず出てくるのがデータ転送処理。

今回は、Treasure Dataが提供しているワークフローエンジンの[digdag](https://www.digdag.io/)とデータ転送ツールの[embulk](https://www.embulk.org/)を用いて、Google Cloud PlatformのGoogle Compute Engine(GCE)インスタンスに、ETL基盤を構築するまでの過程をまとめてみました。
あくまで個人の勉強用の規模のため、実務で利用する場合はセキュリティなどを考慮していただく必要があることをご了承ください。

# 構成図とゴール

Container-Optimized OSのGCEインスタンスを立て、その中でdigdagとembulkのコンテナを起動させ、同じプロジェクト内のGoogle Cloud Storage(GCS)に保存しているCSVファイルをもとに、BigQueryにテーブルを生成するところまでをゴールとします。

GCEのContainer-Optimized OSは、予めコンテナの実行に適した環境を用意してくれているため、サクッと試すにはとても便利なOSです。

![無題のプレゼンテーション.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/311026/641d93a7-64e4-c5c6-29de-2e3ff3e20f92.png)

# GCEのインスタンスを作成

基本的にはデフォルトの設定をいじらず、費用もかけたくなかったため[こちら](https://cloud.google.com/free/docs/gcp-free-tier/?hl=ja#compute)の情報をもとにインスタンスを作成しました。

| 項目 | 値 |
| :-- | :-- |
| 名前 | etl-practice-instance |
| リージョン | us-central1(アイオワ) |
| シリーズ | E2 |
| マシンタイプ | e2-micro(2 vCPU、1GBメモリ) |
| OS | Container Optimized OS |
| バージョン | Container-Optimized OS 85-13310.1453.1 LTS |
| ブートディスクの種類 | 標準永続ディスク |
| サイズ | 10 |


# ローカルからssh接続

ローカルPCにCloud SDKをインストールし、gcloudコマンドを使って先程作ったインスタンスにsshで接続します。
※ブラウザウィンドウでも接続はできるので、この辺はお好みで。

```shell
$ gcloud compute ssh --zone "us-central1-a" "etl-practice-instance"  --project "{{GCPのプロジェクト名}}"

# 作業用のディレクトリを作成しておく
$ mkdir etl-practice
$ exit
```

# サービスアカウントの作成

embulkを通してGCSからBigQueryにデータを転送するため、それぞれのサービスを参照する権限を持つサービスアカウントを作成しておきます。
本来は必要最小限の権限を付与するべきですが、今回は勉強用のため、サービスアカウントに対してIAMにて`ストレージ管理者`と`BigQuery管理者`の権限を付与しました。

また、embulk実行時に認証で必要になるため、JSON形式で認証鍵をダウンロードしておきます。


# 必要なファイルを揃える

GCEインスタンス内で作業をしてもいいのですが、キー入力後のレスポンスが若干遅いので、先にローカルで必要なファイルを揃えた上で、GCEインスタンスに転送することにします。

## サンプルCSV

```csv
id, name
1, taro
2, jiro
3, saburo
```

作成したら、GCSにアップロードしておきます。


## Dockerfile
```dockerfile
FROM openjdk:8-jre-alpine

# embulk
RUN wget -q https://dl.embulk.org/embulk-latest.jar -O /bin/embulk \
  && chmod +x /bin/embulk
RUN apk add --no-cache libc6-compat \
  && embulk gem install embulk-input-gcs \
  && embulk gem install embulk-output-bigquery

# digdag
RUN wget -q "https://dl.digdag.io/digdag-latest" -O /bin/digdag
RUN chmod +x /bin/digdag
RUN echo 'export PATH="$HOME/bin:$PATH"' >> ~/.bashrc
RUN source ~/.bashrc

WORKDIR /work
```

余談ですが、筆者はDockerを全く触ったことがなかったため、[こちら](https://www.amazon.co.jp/%E4%BB%95%E7%B5%84%E3%81%BF%E3%81%A8%E4%BD%BF%E3%81%84%E6%96%B9%E3%81%8C%E3%82%8F%E3%81%8B%E3%82%8B-Docker-Kubernetes%E3%81%AE%E3%81%8D%E3%81%BB%E3%82%93%E3%81%AE%E3%81%8D%E3%81%BB%E3%82%93-%E5%B0%8F%E7%AC%A0%E5%8E%9F%E7%A8%AE%E9%AB%98/dp/4839972745)の書籍をもとに勉強しました。
すごく丁寧に解説されており、これからDockerを触る人にはオススメの1冊です。

## config.yml

```yml
in:
  type: gcs
  bucket: {{CSVファイルをアップロードしたGCSのバケット名}}
  path_prefix: {{拡張子を除いたCSVファイル名}}
  auth_method: json_key
  json_keyfile: {{サービスアカウントのJSON鍵ファイル}}
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    header_line: true
    columns:
      - {name: id, type: long}
      - {name: name, type: string}


out:
  type: bigquery
  mode: replace
  auth_method: service_account
  json_keyfile: {{サービスアカウントのJSON鍵ファイル}}
  project: {{GCPのプロジェクト名}}
  auto_create_dataset: true
  dataset: etl_practice
  table: sample_data
```

各プラグインの詳細は、下記をご参照ください。
[embulk-gcs-input](https://github.com/embulk/embulk-input-gcs)
[embulk-bigquery-output](https://github.com/embulk/embulk-output-bigquery)


## etl.dig

```digdag
timezone: "Asia/Tokyo"

+task1:
  sh>: embulk run config.yml
```

今回の挑戦で初めてdigdagを(というかワークフローエンジンを)触りました。
この辺は、まだまだ勉強中です。。。


# ローカルからGCEに、ファイルをアップロード

`gcloud compute scp`コマンドを利用して、先程作成したファイルをGCEインスタンスにアップロードします。
コマンドに関する詳しい情報は、[こちら](https://cloud.google.com/sdk/gcloud/reference/compute/scp?hl=ja)をご参照ください。

```shell
# 転送するファイルは
# Dockerfile
# config.yml
# etl.dig
# サービスアカウントのJSON鍵ファイル
$ gcloud compute scp --project "{{GCPのプロジェクト名}}" --zone "us-central1-a" --recurse {{ファイルパス}} {{GCPのユーザー名}}@etl-practice-instance:/home/{{GCPのユーザー名}}/etl-practice
```

# コンテナを起動してETL処理を実行

必要な準備がすべて完了したため、ここからは実際にコンテナを起動させ、GCSからBigQueryへのETL処理を実行してみます。


```shell
# ローカルからGCEインスタンスにsshで接続
$ gcloud compute ssh --zone "us-central1-a" "etl-practice-instance"  --project "{{GCPのプロジェクト名}}"

# 作業用ディレクトリに移動
$ cd etl-practice

# Dockerfileをもとに、コンテナイメージを作成
$ docker build -t etl-image .

# コンテナイメージが作成されているかを確認
$ docker image ls

# コンテナイメージをもとに、コンテナを起動
$ docker run -it -v $PWD/:/work etl-image

# digdagを実行
$ digdag run etl.dig
```

BigQueryにアクセスし、`etl_practice`データセットに`sample_data`テーブルができていれば成功です。
お疲れさまでした！

# おまけ

ワークフローエンジンなのに手動実行で終わってしまってはなんだか味気ないので、`etl.dig`ファイルを修正して簡易的なスケジュール実行をしてみます。

## etl.dig


```digdag
timezone: "Asia/Tokyo"

schedule:
 # 例えば毎日AM01:30に実行させたければ、 30 1 * * *
 cron>: {{cron形式でスケジュール実行させたいタイミングを記載}}

+task1:
  sh>: embulk run config.yml
```

```shell
$ digdag scheduler
```

cronで指定したタイミングになると、digdagがetl.digファイルのタスクを実行してくれます。


# おわりに

この挑戦をする前の筆者は、digdagやembulkがどんなツールかは理解していたものの、使い方やそれ自体を動かすにはどうすればいいかが全くわからない状態でした。

しかし、データエンジニアを目指す身としてどうしても自力でETL基盤を構築してみたく、他社様のデータ基盤の事例を拝見しながら、Dockerで動かせることを知り、Dockerの基礎を身に着け、digdagとembulkの使い方を学び、最終的には実際にETL基盤を構築することができました。

実際にそれなりの規模で実装するのであればGoogle Kubernetes Engine(GKE)上で動かしたり、セキュリティやシステム構成などもしっかり考える必要があるなど、まだまだ勉強すべきことは沢山あるため、今回の挑戦で終わるのではなく、これがデータエンジニアになるための第一歩として、これからも勉強を続けていきます。

ここまで読んでくださりありがとうございました！