WorkflowでのJSON処理

WFのHTTPオペレーター

Treasure Data CDPのWFはオープンソースで展開されている「digdag」がベースとなっております。HTTPオペレーターに関してはこちらの資料をご参照ください。

取得元のWebApiとして返却されるJSON例

本サンプルWFでは下記のJSONをHTTPオペレーターで取得します。

取得元のJSONファイル

[
{
"Name":"td_test",
"displayName":"TreasureData_Test",
"createdAt":"2020-12-24T05:30:54Z",
"updatedAt":"2020-12-24T05:30:54Z",
"idField":"fieldId",
"dedupeFields":["sample_field"],
"searchableFields":[
["sample_field"]
],
"Relationships":[
{"field":"fieldid",
"Type":"child",
"relatedTo":{
"name":"test",
"field":"Id"}
}
]
}

]

WFのHTTPオペレーターでデータを取得

WFソースは以下のようになります。

+get_data:
  http>: https://example.com/data.json
  method: GET

  headers:
  - Authorization: 'Bearer *access_token*'

  - content-type: 'application/json'
  store_content: true
  
+empty_tmp_table:
  td_ddl>: 
  empty_tables: [tmp_table]
+insert_d:

  td>:
    data: select '${JSON.parse(http.last_content).result}' as d;
  insert_into: tmp_table
  engine: presto

処理内容としては下記となります。
5-6行目：認証処理はheadersに書き込んでGETするパターン。「Authorization」項目に書き込み
8行目：「content-type」は「’application/json」を指定
16-19行目：一時テーブルにデータを暫定的に格納

そして、取得したJSONをWFでParseする為にJSON_EXTRACTとJSON_EXTRACT_SCALARを利用しましす。（https://prestodb.io/docs/current/functions/json.html）

+parse_json:
  td>:
  query: |
    WITH dataset AS (
      select d as blob from tmp_table
    )  
    , t0 AS
    (
      SELECT

        CAST(JSON_EXTRACT(blob,'$')AS ARRAY) AS hits
      FROM
        dataset
    )
    , t1 AS
    (
      SELECT

      json_format(JSON_EXTRACT(hit,'$.searchableFields')  as searchableFields
      ,JSON_EXTRACT_SCALAR(hit,'$.name') as api_name
      ,JSON_EXTRACT_SCALAR(hit,'$.displayName') as displayName
      FROM
        t0
      CROSS JOIN UNNEST(hits) AS t(hit)
   
  insert_into: json_data
  engine: presto

ここでの処理内容は下記となります。
11行目：「JSON_EXTRACT」を使用してJSONの大本のノードを取得(その場合「$」を指定、JSON型にCASTする必要もあります)
19-24行目：ノードに紐付けJSONキーを指定して値を取得
26-27行目：データをテーブルに格納

まとめ

Treasure Data CDPのワークフローはJSONのパースなど今までプログラミング言語でしかできなかった実用的な機能も保持しております。他にSQL内でのWindow関数など便利な機能もありますのでそちらをご覧いただき、業務にご活用いただければと思います。