ブログ

Blog

絞り込んで表示する

2020.07.30 - 

GCS トリガーで BigQuery へデータを自動取り込みするフローを作る

みなさん、こんにちは。
Groovenauts コンサルタント の藤本です。
今日は MAGELLAN BLOCKS の「フローデザイナー」について新機能のご紹介です。

フローデザイナーとはブロックを繋ぎ合わせるだけで、さまざまなデータを処理するフローを作成できるサービスです。
そんなフローデザイナーに新しくできた Google Cloud Storage（以下 GCS）トリガー機能をご紹介したいと思います。
どういう機能かというと、GCS にファイルがアップロードされたら、自動的にフローを実行してくれる機能なんです。

今回は、とあるデータを BigQuery（以下 BQ）へ取り込むフローを作成し、GCS トリガーで自動実行させてみたいと思います。

1. データを BQ へ取り込むフローを作成

今回はサンプルとして用意した、日毎の来店客数データを BQ へ取り込んでみたいと思います。
データはこんな感じです。

raiten.csv

それでは早速、フローデザイナーでフローを作成してみましょう。

フローデザイナー

フローデザイナー

  1. フローをドラッグ&ドロップで繋げる
    • 今回使うブロックは「BigQuery」カテゴリの以下の2つです。
      • GCSから単一テーブルへロード: CSV ファイルのデータを BQ へ取り込む
      • DataEditor へ登録 : BQ へ取り込んだデータを確認するため DataEditor へ登録する
    • 「フローの開始」→「GCSから単一テーブルへロード」→「DataEditor へ登録」→「フローの終了」の順番にドラッグ&ドロップで繋げます。
  2. 各ブロックの設定を行う
    • 「GCSから単一テーブルへロード」の設定内容
      • GCPサービスアカウント :  MAGELLAN BLOCKS に登録済みの GCP サービスアカウント
      • 投入データのファイルGCS URL : 取り込み元の GCS パス
      • 投入先のデータセット : 取り込み先の BQ のデータセット名
      • 投入先のテーブル : 取り込み先の BQ のテーブル名
      • スキーマ設定 : 各カラム名と型を定義
      • (拡張設定) 空でないテーブルが存在したとき : 「上書き」
      • (拡張設定) 読み飛ばし行数  : 「1行」
    • 「DataEditor へ登録」の設定内容
      • GCPサービスアカウント :  MAGELLAN BLOCKS に登録済みの GCP サービスアカウント
      • 名前 : DataEditor への登録名
      • データセット : 取り込み先の BQ のデータセット名
      • テーブル : 取り込み先の BQ のテーブル名

設定したら保存を忘れずに行ってくださいね。
これで、データを BQ へ取り込むフローが出来ました。

2. GCS トリガーの設定

次に、今回のメインである GCS トリガーの設定を行います。
フローデザイナー画面右上の「フローリスト」をクリックしてフローリストを開きます。
フローリストから、対象のフローの「外部実行」の「設定する」をクリックします。

GCSトリガー

設定画面で以下の設定を行います。

  • フローID : 任意の ID
  • Google Cloud Storage (GCS) にファイル/フォルダを配置した時にこのフローを実行する : チェックを入れる
  • GCPサービスアカウント: MAGELLAN BLOCKS に登録済みの GCP サービスアカウント
  • バケット : トリガーを設定するバケット名

GCSトリガー

設定が終わったら「設定」ボタンを押下し、ステータスが「有効」になるまで少し待ちます。

GCSトリガー

ステータスが「有効」になったら、フローは完成です。
いよいよ、自動実行の動作確認です。

3. GCS へファイルをアップロード

それでは、GCS へファイルをアップロードして、フローが自動実行されるか確認したいと思います。
GCS へのアップロードは GCS Explorer から行います。
トリガー設定したバケットを開き、画面右上のファイルアイコンから「ファイルアップロード」を使って対象のファイルをアップロードします。

GCSアップロード

さて、フローは実行されているでしょうか。
フローデザイナーに戻って確認してみましょう。
フローのログを見てみるとステータスが「実行中」となっており、フローが自動実行されたことがわかりますね。

フローデザイナー

フローのステータスが「成功」になれば、BQ への取り込みが完了です。
ちなみに、GCS へアップロードしてから、BQ へデータ取り込み完了するまでわずか 7 秒で完了しました。さすが、GCP 速いですね！

フローデザイナー

最後に、BQ へデータが取り込みされているかも確認しておきましょう。
BQ のデータの確認はフローで作成した DataEditor を使います。
DataEditor のデータ一覧を開くと「raiten」ができていますね。

DataEditor

「raiten」をクリックして中身を見てみると、データがちゃんと登録されているのがわかります。
下記は DataEditor のサマリ画面になりまして、登録したテーブルのサマリ情報が一目で分かるようになっています。

DataEditor

さいごに

いかがでしたでしょうか。
MAGELLAN BLOCKS ではさまざまなフローを手動で実行するのはもちろん、時間指定で定期実行したり、今回のようにファイルがやってきたらその瞬間に自動で実行することもできます。
設定はとても簡単なので、皆様の業務に合わせて是非活用してみてください。

それでは、最後までお読みいただきありがとうございました！

※本ブログの内容や紹介するサービス・機能は、掲載時点の情報です。