Application Integrationっていう、ETLとかに使えるらしいプロダクトが追加されたのは知っていたが、実際に触ったことは無かった。
Cloud Workflowsと組み合わせて、サーバーレスで色々できるらしい。
ということで今回は、「Google Cloud StorageにアップロードされたJSONファイルを取得し、そのプロパティを抽出して、後段の処理で使うための変数に割り当てる」処理のサンプルを残しておきます。

構築の流れ

Google Cloudプロジェクトで、Application Integration APIを有効化する
Integration(フローの管理単位)を新規作成する
Integrationの中身(フロー)をGUIで作る
フローの詳細パラメーターをJSONで与える
テスト実行

この記事では「3」以降のみ説明します。

実際の手順

今回、値を読み出すファイル upload.txt の中身は、以下のようなJSONです。

{
  "key": "value",
  "today": "Friday"
}

このファイルがGCSバケットにアップロードされているので、それを読み出して、A.I.の変数に割り当てます。

いきなり完成図ですが、向かって右側がフローを追加するフィールド。
向かって左側が、定義した変数が並ぶフィールドです。
フロー(四角いやつ)をクリックすると、右側に詳細設定パネルが出てきます。
フローは、必ずトリガーではじまり、外部から情報を取得するコネクターや、情報を変化させるタスクを経由して進んでいきます。
制御構造については今回は触れません。

GCSバケット内のJSONファイルを取得したいので、Cloud Storageコネクターを追加します。
すると、左ペインにローカル変数「connectionInputPayload」が追加されます。
コネクター自体のプロパティにも、GCPプロジェクトやリージョンなどの設定値を与えますが、バケット名、バケット内のオブジェクト名といった具体的な指定は、この変数でおこないます。
パッと見ではわけが分かりませんが、中ほどにグレーアウトして表示されているのが「スキーマ」のJSONです。
ここに書かれているルールに従って、「Default Value」欄にJSONを書きます。
ここでは、バケット名、ファイル名だけを書いています。

次に、「Data Mapping」です。
読んで字のごとく、データを変数に割り当てる処理です。
GCSコネクターのドキュメントを読むと書いてあるのですが、コネクターを実行すると、

Success
Content

の2つの値をJSONで返してきます。
今回の場合は、以下のような内容です。

[
  {
    "Success": "True",
    "Content": "{\n \"key\": \"value\",\n \"today\": \"Friday\"\n}\n\n"
  }
]

要素数が1個の配列の中に、キーが2つあるハッシュ(オブジェクト)が入っています。
ここで注意してほしいのは、Contentの値は文字列であってオブジェクトではない、ということです。
ここを変換するために、次段の処理が必要になります。

Data Mapping Editorの画面です。
変数、Input、Outputという項目があります。
使うのはInput、Outputだけです。
画面に記述しているように、Inputでは変数を受け取って、その変数に対して何らかの処理をし、Outputの変数に代入します。
ここでは、GCSコネクターの出力からJSON文字列を切り出して、変数JSONContentに割り当てています。

先述した通り、文字列でしかないJSONをオブジェクトとして認識させるため、 .TO_JSON() 関数を適用します。
JSONに変換したので、プロパティである .today にアクセスできます。

テスト実行画面です。
GUIのツールバーから「TEST」を押すとこの画面になります。
何も変更せず TEST INTEGRATION しましょう。

テスト実行結果の画面です。
変数 ChildJSONContent に、入れ子になっていたJSONの値(テキストファイルに書かれていたJSONの値)が取得できています。