Cloud Data FusionでGoogle Cloud StrageのCSVをBigQueryに格納

あまりQiita記事執筆は慣れてないのでご愛嬌ということで、、

#Cloud Data Fusionとは
Googleの公式リファレンスによると、
>Cloud Data Fusionは、データパイプラインを素早く構築、管理できる、クラウドネイティブのフルマネージド エンタープライズ データ統合サービスです。
(引用：https://cloud.google.com/blog/ja/products/data-analytics/set-up-a-secure-data-pipeline-easily-in-the-cloud)

一言で言うと、下図のようにGUIベースでパイプライン構築ができる非常に便利なサービスです。
![Cloud Data fusion image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/8be56b08-a211-f72d-007e-81aec638f7ec.png)
（2021/08 現在、正直バグが多い気がする。今後の発展に期待です）

今回はGoogle Cloud Strage(以下：GCS)などに格納されているCSVを、BigQuery(以下：BQ)にデータを格納するパイプラインをノーコードで構築します。

#前提
 - GCPアカウント作成
 - Cloud Data Fusionのインスタンス作成
 - BQに格納したいcsvデータをGCSにアップロード

未作成の方はこちら→https://cloud.google.com/data-fusion/?hl=ja

#パイプラインの作成

###①データクレンジング
持ってきたデータからBQでの解析に必要なカラムだけに抽出します。

TOP画面で「Wrangle」を選択
![cloud data fusion TOP.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/7cb346dd-945c-6d27-e0c1-595037ff9729.png)

GCSにアップロードしておいたCSVを選択すると、以下のようなテーブルが現れます。
（今回はWordpressで構築したサイトのログデータを持ってきました。１行目はヘッダー）
![スクリーンショット 2021-08-27 11.18.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/fdd2a82d-1c33-da50-b698-f5c9d4a4d8c6.png)

CSVにパースします（元データはCSVのはずですが、なぜかテキスト扱いになっている？？）
![スクリーンショット 2021-08-27 11.23.17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/8e82e031-4e60-cd59-f2d0-6ad24d62798a.png)

「comma」を選択し、１行目がヘッダーの場合は「Set first row as header」をチェックし、Applyをクリック（ファイルに合わせて適宜変換かけてください）
![スクリーンショット 2021-08-27 11.30.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/6d64b1da-b11a-b820-03c3-80e5841a965d.png)

すると綺麗にカラムごとに分かれてくれます。
![スクリーンショット 2021-08-27 11.33.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/dfabf52a-e14d-d359-5671-79ebdb278689.png)

１列目のbody列はもう不要なので削除します。
![スクリーンショット 2021-08-27 11.34.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/1b593133-d4b6-77d3-f683-05c3c359abf4.png)

パースした直後はID列など、数値しか入っていないカラムもString型になっているのでInteger型にキャストします。
![スクリーンショット 2021-08-27 11.37.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/6dea8b97-f63e-7b21-c1bc-a1c5da137ac7.png)

他にも、大文字変換や空文字変換などが可能です（ここでは説明を割愛）

###②パイプライン作成
①が完成したら、「Create a Pipeline」を選択。
![スクリーンショット 2021-08-27 11.40.03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/f57d7b8c-abdd-cf4e-e8b0-9422c3c6c5cf.png)

Batch Pipelineを選択
![スクリーンショット 2021-08-27 11.43.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/123fc113-25fb-5dca-1d33-54810dcdb756.png)

すると、GCSからCSV変換までのパイプラインを自動的に構築してくれます。
![スクリーンショット 2021-08-27 11.44.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/3245baca-6951-6cfa-b993-0d229d84d6d5.png)

※「Wrangler」>「Properties」を選択し、「Output Schema」で出力するカラムの指定ができます。
![スクリーンショット 2021-08-27 11.51.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/1a2044fa-2928-e3e2-225d-b7eb4365773e.png)

###③BQへの接続設定
サイドメニュー>Sinkから「BigQuery」を選択。
![スクリーンショット 2021-08-27 11.45.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/e53b7659-7844-9f3c-e12f-2ebc0ca1c253.png)

「Wrangler」から「BigQuery」まで矢印を伸ばし、Propertiesを選択します。
![スクリーンショット 2021-08-27 11.47.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/5ae5d912-a5c9-50bf-9128-89bcb0374862.png)
![スクリーンショット 2021-08-27 11.49.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/099ae1ba-bb00-2e33-8e7b-1cc597678cf1.png)

必須項目の「Label」「Reference Name」「Dataset」「Table」を埋め、Validateを選択し、「No error found.」と出ればOK。
（※Tableは小文字じゃないと怒られます、怒られた後修正しても怒られた時のデータが消えないバグあり 2021/08/27現在）
![スクリーンショット 2021-08-27 11.51.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/5dd08a94-d83c-b4a5-fd3e-e77c0433995b.png)

###④パイプライン実行
ヘッダーメニュー>Previewから「Run」を選択。Succeedと出ればOK
![スクリーンショット 2021-08-27 11.58.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/15c8baa0-adb5-84f2-f574-8505d962a830.png)

※エラーを吐いた場合は、「Wrangler」>Error Handlingで「Skip error」にして、半強制的にエラーを潰せます（エラー潰してデプロイ実行した後、BQ上の完成品を見ても、ぶっちゃけ問題なさそうだった）
![スクリーンショット 2021-08-27 12.02.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/92a17b95-5c7b-ec0f-49d4-ac7e37ead6d5.png)

Previewをパスしたら、Deployを選択し、
![スクリーンショット 2021-08-27 12.05.44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/a2d2a96c-2900-103e-3abe-ea254e940aaa.png)

Runを選択します。１０分以上は大体かかるので、成功をお祈りします
![スクリーンショット 2021-08-27 12.06.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/9331138e-3558-25dd-1b32-0c705e82268c.png)

StatusがSucceededとなれば成功です。BQの中身を覗きにいきましょう。
図のようにcsvの内容がBQに反映されていれば完成です。
![スクリーンショット 2021-08-27 12.37.09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1046008/aabd208a-39e9-3fe5-4086-af2e56e25c9f.png)

#まとめ
ノーコードで、わかりやすくGCP内の各サービスにデータの受け渡しができるのは魅力的ですね。
今後はオンプレDBやCloud SQL上のDB、AWSのS3からのデータ取り込みしようかなと思います。