Ingestlyで機種判定 / FastlyログをDataflowでリアルタイムに加工する

Fastlyがログの宛先としてGoogle Cloud Pub/Subに対応（2020年4月19日時点ではLimited Available）するということで、Ingestlyで計測したログをリアルタイムに扱う実験をして記事にしました → [FastlyからCloud Pub/Subにログを直送してDataflowで扱うと幸せになれるというお話](https://qiita.com/hsano/items/ec0faf8a93fef7a32155)

今回はIngestlyのログをリアルタイムに加工して、その結果をBigQueryに書き込む処理をDataflowのジョブとして登録してみます。
具体的には、以下の処理を実現してみます。

<img width="1728" alt="スクリーンショット 2020-04-19 13.57.42.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/a1f77b23-e74e-10b4-481a-273a8e750e7d.png">


## 準備

### 前提

- 既にCloud Pub/Subにトピックが作成済みである（例： `ingestly` ）
- 既にFastlyのLoggingで 上記Pub/Sub トピックへのログ送信が構成済みである（かつログにUser-Agentが含まれる）


### 以下のリソースを作成する

Cloud Storageに今回作成するDataflowジョブ用のバケットを作成する（例： `ingestly-files` ）

<img width="841" alt="バケットを作成" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/298f4bda-d272-97b1-9385-a45374677e5c.png">


上記バケットの中に2つのフォルダ `temp` と `staging` を作成する

<img width="873" alt="tempとstagingを作成する" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/e0ea9295-3dc7-943c-0022-907bf9060d7e.png">


Cloud Pub/Subに、既存のトピックに対応するサブスクリプションを作成する（例： `dataflow` ）

<img width="830" alt="トピックに対応するサブスクリプションを作成する" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/2fbbe4b1-bff7-7a11-a917-3bb7b426c9a7.png">

Dataflowで作業環境としてノートブックインスタンスを作成する（例： `ingestly` ）

<img width="820" alt="ノートブックインスタンスを作成する" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/1cd70f1e-5d9f-1246-80ac-1a63dd4792d9.png">

BigQueryに宛先となるテーブルを作成する。
テーブルスキーマは[Ingestly-Dataflowリポジトリのこちらのファイル](https://github.com/ingestly/ingestly-dataflow/blob/master/enrichment_useragent_sample/table_schema)を使う。

`timestamp` フィールドでパーティションするのを忘れずに。

<img width="730" alt="宛先となるテーブルを作成する" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/d31bf13d-234e-a4dc-79b4-daff410c5bae.png">


## Notebookでの作業

今回は一時的な作業環境として使いやすいため、Dataflow内のNotebookを利用します。
もちろんローカルのJupyterのように、Dataflowの外でも同じ作業はできますが、モジュールや認証周りの整備が必要です。

インスタンスが起動したら、Jupyterに入ります。

<img width="520" alt="Jupyterを開く" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/27a0626b-1ad0-7ded-c000-f96779fe4028.png">


### 必要なモジュールをインストールする

メニューで「File」→「New」→「Terminal」を開きます。
開いたターミナルで以下のコマンドを実行し、作業場所を移動＆今回必要なモジュールをインストールします。

```sh
cd /root/notebook/workspace
pip install google-cloud-pubsub
pip install apache-beam
pip install apache-beam[gcp]
pip install PyYAML ua-parser user-agents
```

<img width="902" alt="ターミナルでの作業" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/1aac2ae4-d46f-c261-858d-3a2fca04737f.png">


### 必要なファイルを作成する

メニューで「File」→「New」→「Text File」を開くと新規テキストファイルが作成できるので、2ファイルを作成します。
（作成直後は `untitled.txt` というファイル名なので、右クリックしてRenameします）

1つは[ingestly_enrichment.py](https://github.com/ingestly/ingestly-dataflow/blob/master/enrichment_useragent_sample/ingestly_enrichment.py)で、ジョブを構築する本体のPythonファイルです。

もう一つが[requirements.txt](https://github.com/ingestly/ingestly-dataflow/blob/master/enrichment_useragent_sample/requirements.txt)で、外部モジュールをDataflowのジョブで扱う場合に必要になるものです。

<img width="1066" alt="2つのファイルを作成" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/5542076d-3be2-1530-9292-ebb5ee0fb426.png">

### 変数に必要情報をセットする

`ingestly_enrichment.py` の前半に以下の様な行があるので、それぞれ値を設定します。

```python
# Settings
PROJECT = 'GCP_PROJECT'
SUBSCRIPTION = 'PUBSUB_SUBSCRIPTION'
REGION = 'GCP_REGION'
BUCKET = 'GCS_BUCKET'
DATABASE = 'BQ_DATABASE'
TABLE = 'BQ_TABLE'
```

|変数|意味|例|
|:----|:----|:----|
| PROJECT | GCPのプロジェクト | `ingestly-prod` |
| SUBSCRIPTION | Cloud Pub/Subで作成したサブスクリプション | `dataflow` |
| REGION | GCPのリージョン | `asia-northeast1` |
| BUCKET | Dataflow関連のファイルを保管するGCSバケット | `ingestly-files` |
| DATABASE | データ格納先のBigQueryデータベース | `ingestly` |
| TABLE | データ格納先のBigQueryテーブル | `log_enriched` |

設定が済むとこのようになります。

<img width="262" alt="設定例" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/9bf37051-5d01-e993-e16a-fb22f56b1d4a.png">


### スクリプトを実行しジョブを登録する

Notebookのターミナルに戻り、以下のコマンドでスクリプトを実行するとDataflowにジョブが作成されます。

データの加工に使うモジュールを、 `enrichment` 関数の中でimportしている場合は問題ないのですが、
もしモジュールを先頭でimportしている場合は `--save_main_session` オプションを付ける必要があります。

```sh
python ingestly_enrichment.py --machine_type n1-standard-1 --job_name ingestly-enrichment
```

<img width="1191" alt="スクリプトを実行してジョブを登録" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/354ef732-ab1e-c606-5b4a-a7eeb4dc2147.png">


Dataflowの「ジョブ」ページでジョブを確認します。

<img width="1002" alt="作成されたジョブ" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/43cb9aa3-2652-65a6-a4c7-91f54b11bc16.png">

そして、処理の流れがこのように表現されているのが見えます。

<img width="276" alt="ジョブ内の処理の流れ" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/2a20e637-af3e-5205-bfe6-a940d276a03c.png">


## 結果を確認する

BigQueryのコンソールで以下の様なクエリーを実行すると…

```sql
SELECT
  er_ua_device_brand,
  er_ua_os_family,
  er_ua_browser_family,
  er_ua_browser_version_string,
  COUNT(DISTINCT ingestly_id) as unique_browsers
FROM
  ingestly.log_enriched
WHERE
  timestamp > '2020-01-01'
GROUP BY
  1,
  2,
  3,
  4
```

今回作成したDataflowを使って、User-Agentから判定したデバイス/ブラウザ情報が確認できます。

<img width="688" alt="スクリーンショット 2020-04-19 13.50.59.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/107873/8e39a9c6-387d-772b-4f8a-cc28d9a9dc4b.png">


## まとめ

ということで、（ニア）リアルタイムなデータ拡張がほとんどコピー＆ペーストで実現できました。
Pythonの処理を発展させることで外部APIの参照等も簡単に実現できるので、データを蓄積するだけではなくビジネスロジックを組み込んで独自のデータ拡張やデータ連携が構築できると思います。


## 注意
- この処理を1ヶ月動かすとGCPの費用が $100/月〜 かかります。
- 不要になったジョブや各種リソースは停止または削除しましょう。
- 最小構成の「n1-standard-1」を選択していますが、データ量に応じてマシンタイプやスケーリングの設定は調整しましょう。