Lookup Catalog 同期ワークフロー

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

GitHub 上のワークフローファイル

最新のワークフローファイルは treasure-data/treasure-boxes リポジトリの 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 回の実行を単一テーブルに限定できます（テスト用途に便利です）。

仕組み

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

変更検知: 各テーブルには対応するダイジェストテーブル（_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 サポートにお問い合わせください。

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

Data Workbench で新しい Workflow プロジェクトを作成します（既存のプロジェクトがある場合はそれを使用します）。
treasure-boxes リポジトリの realtime-box/lookup-catalog-sync/ からファイル（lookup_catalog_sync.dig、scripts.py、queries/discover_tables.sql）をコピーし、プロジェクトにアップロードします。
lookup_catalog_sync.dig を編集し、_export ブロックで reactor_importer_endpoint、reactor_instance、td.database を設定します。テスト用途で実行を単一テーブルに限定したい場合は、オプションで p_table_name も設定します。
プロジェクトの Secrets タブに TD API キーを td.apikey として保存します。フルアクセス権限のキーではなく、必要最小限の権限（Bulk Load API への書き込み権限）に絞ったキーを使用してください。
スケジュールを設定します（毎日の実行を推奨）。設定後、ワークフローを手動で 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`。その他の型は主キーカラムとしてサポートされていません。
一意性	データは重複除去が必要です。ワークフローは行の重複除去を行いません。