# Lookup Catalog 同期ワークフロー

Lookup Catalog 同期ワークフローは、Data Workbench の `cdp_lookup_catalog` データベースから読み取り、ハッシュベースの変更追跡で変更されたレコードを検知し、Bulk Load API を通じて変更された行のみを RT 2.0 の内部ストレージにアップロードします。

GitHub 上のワークフローファイル
最新のワークフローファイルは [treasure-data/treasure-boxes](https://github.com/treasure-data/treasure-boxes/tree/master/realtime-box/lookup-catalog-sync) リポジトリの `realtime-box/lookup-catalog-sync/` に公開されています:

* `lookup_catalog_sync.dig` — メインのワークフロー
* `scripts.py` — テーブルごとに型に応じた JSON ペイロード生成 SQL を作成
* `queries/discover_tables.sql` — `cdp_lookup_catalog` 内の対象テーブルを検出


このワークフローは実行ごとに `cdp_lookup_catalog` 内の対象テーブルをすべて自動検出します。テーブルごとに個別の設定手順は不要です。オプションの `p_table_name` パラメーターを設定すると、1 回の実行を単一テーブルに限定できます（テスト用途に便利です）。

## 仕組み

ワークフローは、検出したすべてのテーブルに対して同じ同期パイプラインを実行します:

```mermaid
flowchart TD
    A[Discover tables] --> B[For each table]
    B --> C[Extract changed records\nvia hash comparison]
    C --> D{Changes exist?}
    D -->|Yes| E[Upload to RT 2.0\nInternal Storage]
    D -->|No| F[Skip — no update]
    E --> G[Update digest table]
    G --> H[Cleanup temp tables]
```

**変更検知**: 各テーブルには対応するダイジェストテーブル（`_wf_{table_name}_digests`）があり、各行の JSON ペイロードのハッシュが保存されています。実行のたびに、ハッシュが変わった行（または新規行）のみがアップロードされます。

**内部テーブルの命名**: ワークフロー内部で使用するテーブルや一時テーブルにはすべて `_wf_` プレフィックスが付き、ソーステーブルと区別されます。`discover_tables.sql` クエリは `_wf_*` テーブルを自動的に同期対象から除外します。

## 設定パラメーター

| **パラメーター**  | **デフォルト値**  | **説明**  |
|  --- | --- | --- |
| `reactor_importer_endpoint` | （必須） | リージョン固有の Bulk Load API エンドポイント。下記のエンドポイント一覧を参照してください。 |
| `reactor_instance` | （必須） | アカウントの RT 2.0 セットアップ時に Treasure AI が払い出すインスタンス識別子（形式: `a{accountId}n{instanceId}`）。Customer Success Manager またはサポートチームにお問い合わせください。 |
| `td.database` | `cdp_lookup_catalog` | ルックアップテーブルを含むソースデータベース。 |
| `p_table_name` | `""` （すべてのテーブル） | 設定した場合、指定テーブルのみを同期します。すべての検出テーブルを同期する場合は空のままにしてください。 |
| `batch_size` | `1000` | Bulk Load API への 1 リクエストあたりのレコード数。 |
| `parallelism` | `10` | テーブルごとの並列アップロードスレッド数。 |


### リージョン別エンドポイント

| **リージョン**  | **`reactor_importer_endpoint`**  |
|  --- | --- |
| US | `https://bulk-storage-importer-api-production-aws-us-east-1.internal.treasuredata.com` |
| EU (eu01) | `https://bulk-storage-importer-api-production-eu01-eu-central-1.internal.treasuredata.com` |
| Tokyo | `https://bulk-storage-importer-api-production-aws-tokyo-ap-northeast-1.internal.treasuredata.com` |


内部エンドポイントへのアクセスについて
`.internal.treasuredata.com` エンドポイントは Treasure Data プラットフォーム内部（例: TD Workflow のタスク）からのみアクセス可能で、パブリックインターネットからはアクセスできません。

インスタンス単位のデータ分離はありません
`reactor_instance` パラメーターはアップロードリクエストのルーティングに使用されますが、データを読み取れるインスタンスを制限するものではありません。アカウント内のすべてのインスタンスが同じ Lookup Catalog ストレージを共有します。つまり、どの `reactor_instance` 値でアップロードされたテーブルであっても、アカウント内の任意のインスタンスの任意の Lookup Catalog Attribute から読み取り可能です。複数の RT 2.0 インスタンスを運用しており、それらの間でカタログデータを共有してはならない場合は、`reactor_instance` による分離に依存せず、アカウントを分けるか Treasure AI サポートにお問い合わせください。

## ワークフローのセットアップ

1. Data Workbench で新しい Workflow プロジェクトを作成します（既存のプロジェクトがある場合はそれを使用します）。
2. treasure-boxes リポジトリの [realtime-box/lookup-catalog-sync/](https://github.com/treasure-data/treasure-boxes/tree/master/realtime-box/lookup-catalog-sync) からファイル（`lookup_catalog_sync.dig`、`scripts.py`、`queries/discover_tables.sql`）をコピーし、プロジェクトにアップロードします。
3. `lookup_catalog_sync.dig` を編集し、`_export` ブロックで `reactor_importer_endpoint`、`reactor_instance`、`td.database` を設定します。テスト用途で実行を単一テーブルに限定したい場合は、オプションで `p_table_name` も設定します。
4. プロジェクトの **Secrets** タブに TD API キーを `td.apikey` として保存します。フルアクセス権限のキーではなく、必要最小限の権限（Bulk Load API への書き込み権限）に絞ったキーを使用してください。
5. スケジュールを設定します（毎日の実行を推奨）。設定後、ワークフローを手動で 1 回実行して、初回アップロードが成功することを確認します。


検出クエリは `cdp_lookup_catalog` 内のすべてのベーステーブルを自動的に検出し、`_wf_*` 内部テーブルを除外します。

並列処理の目安
初回フルアップロードは、デフォルト設定（`batch_size: 1000`、`parallelism: 10`）から開始してください。増分実行時はスロットリングエラーを監視しながら徐々に増やしてください。テーブルサイズが大きいほど、または並列処理数が多いほど、スロットリングのリスクが高まります。

## スケジュールの推奨設定

| **データの種類**  | **推奨スケジュール**  |
|  --- | --- |
| フラッシュセール・時間に敏感な価格情報 | 1〜2 時間ごと（`0 */1 * * *`） |
| 日替わりプロモーション・クーポンテーブル | 6 時間ごと（`0 */6 * * *`） |
| 商品カタログ（安定したデータ） | 毎日（`0 2 * * *`） |


## トラブルシューティング

| **症状**  | **原因と対処法**  |
|  --- | --- |
| 毎回全レコードがアップロードされる | ダイジェストテーブルが存在しないか、ハッシュが古い状態です。`_wf_{table_name}_digests` を削除して再実行し、最初から再構築してください。 |
| 400 エラー: `"Expected string, received..."` | 行に `NULL` または空の主キーがあります。次回実行前にソーステーブルの NULL キーの行をフィルタリングまたは修正してください。 |
| 400 エラー: `"Missing required header: x-lookup-table-name"` | アップロードステップに `x-lookup-table-name` ヘッダーが設定されていません。ワークフローテンプレートのヘッダー設定を確認してください。 |
| `varchar = bigint` 型エラーでワークフローが失敗する | ソーステーブルのカラム型が互換性のない型に変更されています（例: `string` → `int`）。ダイジェストテーブルを削除し、ソースデータを修正して再実行してください。 |
| 主キーカラム変更後にスキーマが一致しない | `_wf_{table_name}_digests` テーブルを削除してから再実行してください。ワークフローは新しいキーで全行を再アップロードします。古いキーでアップロードされたデータは自動的には削除されません。 |
| 削除したカラムがレスポンスに残る | RT 2.0 はアップロードされたデータをそのまま保存します。ソーステーブルからカラムを削除しても内部ストレージのデータは削除されません。残存データを削除するにはフル再アップロードが必要です。ダイジェストテーブル（`_wf_{table_name}_digests`）を削除して再実行してください。それまでは削除済みカラムのデータが API 経由でアクセス可能な状態が続きます。 |
| アップロードがタイムアウトする | `parallelism` または `batch_size` を減らしてください。特に大きなテーブルの初回フルアップロード時に有効です。 |


## キーカラムの制約

ワークフローは、テーブルスキーマの**最初の `time` 以外のカラム**（`information_schema.columns` の `ordinal_position` 順）を主キーとして検出します。テーブルの最初のカラムが `time` という名前の場合は、2 番目のカラムが主キーとして使用されます。

| **制約**  | **詳細**  |
|  --- | --- |
| カラムの位置 | キーカラムはテーブルスキーマの最初の `time` 以外のカラムでなければなりません。 |
| カラム名 | `time` という名前は使用できません（Treasure AI プラットフォームで予約されています）。 |
| NULL 値 | キーカラムに `NULL` または空の値を含めることはできません。NULL キーが存在する場合、アップロードは 400 エラーで失敗します。 |
| サポートされる型 | `string`。その他の型は主キーカラムとしてサポートされていません。 |
| 一意性 | データは重複除去が必要です。ワークフローは行の重複除去を行いません。 |