Cloud Dataflow SQLのデータソースにGCSを使う

この記事は「Google Cloud Platform Advent Calendar 2019」8日目の記事です。

Cloud Dataflow SQLのデータソースにGCSが指定できるようになったので、試してみた話です。

# Cloud Dataflow SQLとは

Cloud Dataflow SQL[^1]は、BigQueryのコンソール画面でSQLを実行することでDataflowのストリーミングジョブを実行できる機能です。

Google Cloud Next '19のセッションでも紹介され、2019/05/16にパブリックα版として公開されました[^2]。

GCPの公式ドキュメント内に[ウォークスルー](https://cloud.google.com/dataflow/docs/guides/sql/dataflow-sql-ui-walkthrough?hl=ja)があります。


# 2019/11/18の変更点[^3]

- Use Cloud Storage filesets as a data source
- Assign schemas to data sources in the Cloud Dataflow SQL UI
- Preview the content of Cloud Pub/Sub messages from the Cloud Dataflow SQL UI

ということで、GCSをデータソースとして使えるようになりました。
使い方のドキュメントは[こちら](https://cloud.google.com/dataflow/docs/guides/sql/data-sources-destinations?hl=en#reading-gcs-fileset)。


アーキテクチャのイメージ。
Cloud Dataflow SQLのデータソースとしてPub/Sub、GCS、BigQueryが指定できるので、Pub/Subから来たデータとGCSやBigQueryにあるデータをくっつけてBigQueryに格納する使い方になります。
![スクリーンショット 2019-12-06 22.29.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/175280/795b0d12-1a99-5c0e-65b6-f04150dd1b0d.png)


# 試してみる

GCS上のファイルに対してCloud Data Catalogでメタ情報（スキーマ定義とか）を登録し、それをクエリで利用します。
Pub/SubをDataflowのソースとして使うときと同様の流れです。

```shell

$ gcloud beta data-catalog entry-groups create test_entry_group --location=us-central1
API [datacatalog.googleapis.com] not enabled on project [62041182717].
 Would you like to enable and retry (this will take a few minutes)?
(y/N)?  y

Enabling service [datacatalog.googleapis.com] on project [62041182717]...
Operation "operations/acf.43376de5-7d2b-4895-bf41-dbb6c56a40a0" finished successfully.
Created entry group [test_entry_group].


$ gcloud beta data-catalog entries create test_fileset \
  --entry-group=test_entry_group \
  --location=us-central1 \
  --gcs-file-patterns=gs://sandbox-suzutatsu-bq-dataflowsql/*.csv \
  --description="テストデータが入ったCSVファイルのファイルセットです。"
Created entry [test_fileset].
```

※ファイルセットには、ヘッダー行のないCSVファイルのみを含める必要があります。

次に、データソースとしてGCSのファイルを使ってみます。
BigQueryのコンソール画面で、`展開`をクリックして`クエリの設定`を選択し、クエリエンジンを`Cloud Dataflow エンジン`に切り替えます。
その後、BigQueryのコンソール画面のナビゲーションパネルにある`データを追加`から`Cloud Dataflow のソース`を選択します。
Cloud Dataflowのソースを追加する際に、ラジオボタンが追加され、GCSのファイルセットが選べるようになっています。
![スクリーンショット 2019-12-06 22.58.06.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/175280/921f6ac4-2331-4cb6-2dcd-213deeab7d38.png)
Cloud Dataflowのソースとして追加して、スキーマを見てみると、未登録の状態です。
![スクリーンショット 2019-12-06 22.59.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/175280/5a0dedf4-ab50-e13e-ca95-c8835f4e6b17.png)
詳細はこんな感じ。
ファイルセットIDにロケーションが入っていたり、タイプがFILESETになっていたりします。
![スクリーンショット 2019-12-06 23.00.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/175280/525eebbc-dcd8-d636-2e7b-fd5cd6be317c.png)
スキーマの設定はGUIでも、JSONを貼り付けてもOK。BigQueryのテーブルを作るときと同じです。
![スクリーンショット 2019-12-07 0.11.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/175280/40cf7929-9618-d4cc-06dc-1448fcebea99.png)


ここまで設定できたら、Pub/Subのデーターソースと組み合わせて、以下のようなSQLクエリを書いて、Cloud Dataflowのストリーミングジョブとして実行できます。

```SQL
SELECT
  a.*,
  b.department_name
FROM
  pubsub.topic.`sandbox-suzutatsu-bq`.sales AS a
LEFT JOIN
  datacatalog.entry.`sandbox-suzutatsu-bq`.`us-central1`.test_entry_group.test_fileset AS b
ON
  a.payload.department_code = b.department_code
```

※なお、クエリをフォーマットするボタンは選択できますが、クエリは変わらずでした。

現時点では、クエリ構文やデータ型に様々な[制約事項](https://cloud.google.com/dataflow/docs/reference/sql/)がありますので、BigQueryの時と同じ気持ちで書くとエラーが出ます。実行時エラーもあります。

# 2019/11/18のその他の変更点

## Cloud Dataflow SQLのUIでスキーマをデータソースに割り当てられるようになった

データソースのスキーマ定義画面に`Edit schema`ボタンが追加され、スキーマの設定が変えられるようになりました。
スキーマ定義が不適切でDataflowジョブが失敗した時に、UIからすぐ直せるのは便利です。

## Cloud Dataflow SQL UIからCloud Pub/Subメッセージのコンテンツをプレビューできるようになった

データソースのスキーマ定義画面に`トピックをプレビュー`というリンクが追加されました。
Pub/Subの`トピックの詳細`画面で`メッセージを表示`をクリックした時と同じ画面が表示されます。

https://cloud.google.com/dataflow/docs/guides/sql/data-sources-destinations?hl=en#gcs

# さいごに

まだα版なので実戦投入はできませんが、SQLクエリでストリーミングデータの加工が書けるのは魅力的なので、今後の機能拡張にも期待したいです。

[^1]: https://cloud.google.com/dataflow/docs/guides/sql/dataflow-sql-intro?hl=ja
[^2]: https://cloud.google.com/dataflow/docs/resources/release-notes-service?hl=ja#May_16_2019
[^3]: https://cloud.google.com/dataflow/docs/resources/release-notes-service?hl=ja#November_18_2019