TROCCOのHTTPS転送でSlack APIのページングを無理やり実現してみた

TROCCOのHTTPS転送ではカーソルベースのページングに対応しており、カーソルが以下のいずれかになるまでリクエストを繰り返します。

* カーソルが含まれていない
* カーソルの値が `null`

> 参考: [転送元 - HTTP・HTTPS](https://documents.trocco.io/docs/data-source-http#%E3%82%AB%E3%83%BC%E3%82%BD%E3%83%AB%E3%83%99%E3%83%BC%E3%82%B9%E3%82%92%E9%81%B8%E6%8A%9E%E3%81%97%E3%81%9F%E5%A0%B4%E5%90%88)

しかし、Slack APIのページングでは最終ページのカーソルに空文字が入ることになっており、HTTPS転送でページングを行うと下記のような感じで無限ループしてしまいます。

```
Request 1: GET https://slack.com/api/conversations.list
Request 2: GET https://slack.com/api/conversations.list?cursor=XXXXXXXX
Request 3: GET https://slack.com/api/conversations.list?cursor=YYYYYYYY
Request 4: GET https://slack.com/api/conversations.list?cursor=ZZZZZZZZ
Request 5: GET https://slack.com/api/conversations.list?cursor=
Request 6: GET https://slack.com/api/conversations.list?cursor=XXXXXXXX
...
```

この問題を無理やり解決してみたので、この記事ではその方法について記します。

## 解決方法

解決方法を図にまとめたものを以下に示します（図はconversations.list APIを叩く場合を例として描かれています）。

この方法のポイントは次の点です。

1. Slack APIのレスポンスをJSON Lines（＝1行のレコード）としてBigQuery上にあるテーブルへ追記
2. TROCCOの転送ジョブ設定で転送日時カラムを設定
3. TROCCOのカスタム変数ループで最後に転送されたレコードの `next_cursor` を取得し、それを使って次のSlack APIリクエストを行う
4. 1.~3.を行うジョブをページ数分だけ複製する

:::note
**（補足）ジョブ複製なしで実現する方法**

4.のジョブ複製をなしで実現する方法もあります。詳細は後述します。
:::

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/ce971835-dd5c-ecf3-9c58-d14af20c96cf.jpeg)

### 具体例

次の要件を満たすワークフローを作成する場合を例に具体的に説明します。

要件:
* [conversations.list（チャンネル一覧を取得するAPIメソッド）](https://api.slack.com/methods/conversations.list) を叩いてBigQuery上に「channels」というチャンネル一覧テーブルを作成する
* channelsテーブルは日次で洗い替える

### 全体像

ワークフローの全体像は次のとおりです（図は4ページまで取得する場合で描かれています）。

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/bbe57caf-a730-f477-a7b2-499de545a0fa.jpeg)

テーブルリネージは次のとおりです。conversations.list APIのレスポンスを一旦conversations_listテーブルへ格納し、その後加工してchannelsテーブルを作成します。

```mermaid
graph LR
    conversations.list_api([conversations.list API])
    subgraph BigQuery
        conversations_list
        channels
    end
    
    conversations.list_api
    --> conversations_list
    --> channels
```

### ① conversations_listテーブル初期化用データマートシンクジョブ

データマートシンクジョブで次のクエリを実行し、conversations_listテーブルを新規作成（すでに存在している場合は初期化）します。

conversations_listテーブルはconversations.list APIのレスポンスを1レスポンス1行で格納していくためのテーブルです。

:::note
**（補足）1レスポンス1行で格納する理由**

Slack APIではレスポンスのJSONオブジェクトの構造が要素（conversations.listの場合、チャンネル）によって微妙に異なることがあるためです。

詳細は次の記事をご覧ください。

> [「内定者インターン時代に半年かけて作った全社横断のSlackデータ基盤、TROCCOなら爆速構築できる説」の検証 - Yappli Tech Blog](https://tech.yappli.io/entry/rebuild-slack-datapipeline-with-trocco#conversationslistconversationslist-API%E5%AE%9F%E8%A1%8C%E7%94%A8%E3%83%87%E3%83%BC%E3%82%BF%E8%BB%A2%E9%80%81:~:text=%F0%9F%92%A1-,%E5%85%A5%E5%8A%9B%E3%83%95%E3%82%A1%E3%82%A4%E3%83%AB%E5%BD%A2%E5%BC%8F%E3%82%92JSON%20Lines%E3%81%AB%E3%81%99%E3%82%8B,-conversations.list%20API)

:::

このジョブのポイントは次のとおりです。

* **②、③で転送日時を格納するためのカラム（今回は `transferred_at` ）を用意する**
* **`transferred_at` に確実に最古なタイムスタンプ（今回は `2000-01-01 00:00:00` ）を設定する**

```sql
CREATE OR REPLACE TABLE `YOUR_PROJECT_ID.YOUR_DATASET.conversations_list`
AS

SELECT
  false AS ok,
  "" AS channels,
  "" AS response_metadata,
  TIMESTAMP("2000-01-01 00:00:00") AS transferred_at, -- 確実に最古なタイムスタンプを設定
  "invalid_cursor" AS `error`
```

### ② conversations.list API初回リクエスト用データ転送ジョブ

conversations.list APIを叩くためのデータ転送ジョブの初回リクエスト用です（初回と2回目以降でジョブを分けている理由については後述します）。

TROCCOでの転送設定は次のとおりです。conversations.list APIのレスポンスを1行のJSON Linesとして①で作成したconversations_listテーブルへAPPENDしています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/111626ec-1e8e-6f11-7cf0-f48d34cbee55.png)

### ③ conversations.list API2回目以降リクエスト用データ転送ジョブ

conversations.list APIを叩くためのデータ転送ジョブの2回目以降リクエスト用です。

TROCCOでの転送設定は下図のとおりです。②のジョブとの違いを以下に記します。

* **カスタム変数 `$cursor$` を設定している**
  → `$cursor$` はカスタム変数ループで次のページへのカーソル値を入れるための変数です
* **conversations.list APIのパラメータに `cursor=$cursor$` を設定している**
  → これにより、 `$cursor$` に `next_cursor` の値をセットすればページ送りができます

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/d61188d5-d736-af2f-b47b-2305815e0a09.png)

このジョブに対して、ワークフロー上で次のカスタム変数ループを設定します。設定しているクエリでは、最後に転送されたレコードの `next_cursor` を `$cursor$` へセットしています（つまり、このカスタム変数ループは1周しかしません）。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/fad0b31e-67a4-4d58-70cb-bc2611c86115.png)

ワークフロー上でこのジョブを複製すれば、複製した数だけページを送ることができます。

:::note
**（補足）初回と2回目以降でジョブを分けている理由**

下記のようにすれば、2回目以降のジョブでも初回のジョブと同じことができるように思えます。

* ①のジョブでconversations_listテーブルを初期化する際に、 `ok` カラムの値に `false` を入れておく
* これにより、次のようにして③の2回目以降リクエスト用ジョブで②の初回リクエスト用ジョブを実現できるはず
    * ③のジョブが初めて動くとき、先ほどのカスタム変数ループのクエリは `ok=false` として動く
    * この際に返す値を空文字にしておけば、リクエスト時に叩かれるURLは次のようになり、初回リクエストが実現できる
      ```
      conversations.list?cursor=
      ```

クエリで説明したものを下記に記します。

```diff_sql
  IF(ok, -- Slack APIの仕様より、レスポンスが正常に取得できていたらok=trueが入っている
    JSON_EXTRACT_SCALAR(response_metadata, "$.next_cursor"),　-- next_cursorを取得
+   "" -- 初回実行時に$cursor$へセットされる値
+      -- 空文字にしておけばconversations.list?cursor=となり、初回用ジョブを2回目以降用ジョブで実現できる
    ) AS next_cursor
```

しかし、実際にやってみると実行時に次のエラーが出てしまいます。どうやらカスタム変数に空文字はセットできないようです。

```
IllegalStateException: params: value or values is required. Suppressed: NullPointerException
```

この問題を回避するため、初回と2回目以降でジョブを分けている次第です。

---

（以下、個人的なコメントです）

転送設定でカスタム変数を設定する際に、値を空で設定できないのはこのためだったんですね…。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/b2063a5d-5f07-3366-b14e-823035d71b2b.png)

ただ、HTTPS転送のカーソルベースのページングでは空文字でいけていたわけなので、下記2つで処理（バリデーション等）が異なるのだと思われます :thinking: 

* HTTPS転送でのカーソル値のセット方法
* カスタム変数によるカーソル値のセット方法

:::

### ④ channelsテーブル作成用データマートシンクジョブ

次のクエリを実行し、conversations_listテーブルからchannlesテーブルを作成します。

```sql
CREATE OR REPLACE TABLE `YOUR_PROJECT_ID.YOUR_DATASET.channels`
AS

WITH

unnest_channels AS (
SELECT
  channel
FROM
  `YOUR_PROJECT_ID.YOUR_DATASET.conversations_list`,
  UNNEST(JSON_EXTRACT_ARRAY(channels, "$.")) AS channel
),

final AS (
SELECT
  JSON_EXTRACT_SCALAR(channel, "$.id") AS channel_id,
  JSON_EXTRACT_SCALAR(channel, "$.name") AS channel_name,
  channel
FROM
  unnest_channels
)

SELECT * FROM final
```

## ジョブ複製なしで実現する方法

カスタム変数ループを利用することで、③の2回目以降リクエスト用ジョブの複製を不要にすることができます。

### 方法

ワークフローの全体像は次のとおりです。

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/c71af91b-d55a-13de-c549-dafa1c50ba5e.jpeg)

ポイントは次のとおりです。

💡 **③子のジョブのカスタム変数に `$loop$` を追加する**

`$cursor$` に加え、 `$loop$` をカスタム変数ループに追加します。 `$loop$` は後述する親ワークフローでのカスタム変数ループで使用します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/fb865abb-7dce-76b1-eb16-ac9f01564105.png)

💡 **③子のジョブをまとめて1つのワークフロー（子ワークフロー）に切り出し、親ワークフローの③親のジョブで `$loop$` を使ってカスタム変数ループさせる**

ワークフロー全体像の下図の部分です。カスタム変数ループのクエリにて、 `$loop$` をループ変数として `GENERATE_ARRAY()` で指定した回数だけループさせる＝③子のジョブを実行させるようにしています。指定する回数はページ数より確実に多くしてください（でないと、最終ページまで到達できません）。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/da655465-b8bb-34f8-2e6c-49d708367273.png)

💡 **最終ページに到達したら無効なカーソル値を返し、わざとジョブを失敗させてループを終了させる**

③子のジョブのカスタム変数ループに設定するクエリを次のようにします。ポイントは、最終ページだったら無効なカーソル値を返すことです。これにより、以降のループで無効なリクエストが発生してジョブが失敗し、ループが終了します。

```diff_sql
SELECT
  IF(ok,
+   IF(JSON_EXTRACT_SCALAR(response_metadata, "$.next_cursor") = "", -- 最終ページかどうかを判別
+     "xxx", -- 最終ページだった場合、ループを止めるために無効なカーソル値を返す
+     JSON_EXTRACT_SCALAR(response_metadata, "$.next_cursor") -- それ以外の場合、next_cursorを取得して返す
+     ),
    ""
    ) AS next_cursor
FROM
  `YOUR_GC_PROJECT_ID.YOUR_DATASET.conversations_list`
ORDER BY
  transferred_at DESC
LIMIT
  1
```

ループが終了するまでの流れをトレースした図を以下に示します（図は3ページ目が最終ページだった場合で描かれています）。

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/05355e64-c37e-f23d-8979-95e6faa3f161.jpeg)

このままだと③親のジョブがエラーとなり、そこで親ワークフローが終了してしまいます。これを回避するため、親ワークフローはエラーが発生しても処理を継続するよう設定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/489030/1a03a0e7-e318-f96e-981a-6590a6658628.png)

### 留意点

この方法だとワークフローの実行が成功していようが失敗していようが完了ステータスがエラーとなってしまいます。そのため、ワークフローのエラーハンドリングが難しくなります。

:::note
**（補足）カスタム変数ループに設定したクエリの実行タイミング**

カスタム変数ループに設定したクエリが実行されるのはループ開始時のみで、ループのたびに実行されるわけではないようです。これは、ループの終了条件に動的な要素を加えることはできないことを意味します。

例えば今回の場合、チャート1のフローチャートのような処理はできません。そのため、チャート2のフローチャートのような処理を行っています。

<table>
<tr>
<td>チャート1<br />（これはできない）</td>
<td>チャート2<br />（だからこうしている）</td>
</tr>

<tr>
<td>

```mermaid
flowchart TD
    start([START])
    fin([END])

    loop_fin_condition{
        next_cursor != ””
        or 
        ERROR is not OCCURRED
        }
    loop_cursor_update[
        **カスタム変数ループ展開**
        **（クエリ実行）**
        $cursor$ = next_cursor
        ]

    request[
        conversations.list
        リクエスト用ジョブを実行
        ]



    start
    --> loop_cursor_update
    --> request
    --> loop_fin_condition
    -- YES --> loop_cursor_update

    loop_fin_condition
    -- NO --> fin
```

</td>

<td>

```mermaid
flowchart TD
    start([START])
    fin([END])

    subgraph 親ワークフロー

        init_parent_loop_item_array[
            **カスタム変数ループ展開**
            **（クエリ実行）**
            parent_loop_array
            = ［1, 2, ..., 10］
            i = 0
            ]
        parent_loop_fin_condition{
            i < len（parent_loop_array）
            or 
            ERROR is not OCCURRED
            }
        parent_loop_loop_update[
            $loop$
            = parent_loop_array［i］
            ]
        parent_loop_i_update[i += 1]
    
        subgraph 子ワークフロー
    
        init_child_loop_item_array[
            **カスタム変数ループ展開**
            **（クエリ実行）**
            child_loop_array =［next_cursor］
            j = 0
            ]
        child_loop_fin_condition{
            j < len（child_loop_array）
            or 
            ERROR is not OCCURRED
            }
        child_loop_cursor_update[
            $cursor$
            = child_loop_array［j］
            ]
        child_loop_j_update[j += 1]
    
        request[
            conversations.list
            リクエスト用ジョブを実行
            ]
        end
    
    end



    start
    
    --> init_parent_loop_item_array
    --> parent_loop_loop_update

    --> init_child_loop_item_array
    --> child_loop_cursor_update
    
    --> request

    --> child_loop_j_update
    --> child_loop_fin_condition
    
    child_loop_fin_condition 
    -- YES --> child_loop_cursor_update

    child_loop_fin_condition
    -- NO --> parent_loop_i_update
    --> parent_loop_fin_condition
    -- YES --> parent_loop_loop_update
    
    parent_loop_fin_condition -- NO --> fin
```

</td>

</tr>
</table>

:::




## 結び

この記事では、TROCCOのHTTPS転送を使ってSlack APIのページングを無理やり行う方法について記しました。

書いておいてなんですが、ここまでするんだったら素直にコード書いてスクリプトを用意した方が運用しやすいんじゃないかなと思います…。ただ、今回採った「レスポンスを一旦JSON LinesとしてAPPENDして後で整形する」という方法はスクリプトを用意する場合でも有用かなと思います（つまり、Slack APIを叩くのをTROCCOでやるか自作スクリプトでやるかの違いです）。

今回の取り組みで、普段なら直面しないであろうTROCCOの細かい仕様に触れることができてとても面白かったです！

ここまでお読みいただきありがとうございました :bow: