The Lookup Catalog sync workflow reads from the cdp_lookup_catalog database in Data Workbench, detects changed records using hash-based change tracking, and uploads only modified rows to RT 2.0's internal storage via the Bulk Load API.
The latest workflow files are available in the treasure-data/treasure-boxes repository under realtime-box/lookup-catalog-sync/:
lookup_catalog_sync.dig— main workflowscripts.py— generates type-aware JSON payload SQL per tablequeries/discover_tables.sql— discovers eligible tables incdp_lookup_catalog
The workflow automatically discovers all eligible tables in cdp_lookup_catalog on each run — there is no separate configuration step per table. An optional p_table_name parameter can restrict a single run to one table, which is useful for testing.
The workflow follows the same per-table sync pipeline for every discovered table:
Change detection: Each table has a companion digest table (_wf_{table_name}_digests) that stores a hash of each row's JSON payload. On each run, only rows whose hash has changed (or that are new) are uploaded.
Internal table naming: All workflow-internal and temporary tables use a _wf_ prefix to distinguish them from source tables. The discover_tables.sql query automatically excludes _wf_* tables from sync.
| Parameter | Default | Description |
|---|---|---|
reactor_importer_endpoint | (required) | Region-specific Bulk Load API endpoint. See the region-specific endpoints table below. |
reactor_instance | (required) | Instance identifier provisioned by Treasure AI when RT 2.0 is set up for your account (format: a{accountId}n{instanceId}). Contact your Customer Success Manager or Treasure AI Support to obtain this value. |
td.database | cdp_lookup_catalog | Source database containing the lookup tables. |
p_table_name | "" (all tables) | When set, syncs only the specified table. Leave empty to sync all discovered tables. |
batch_size | 1000 | Number of records per upload request to the Bulk Load API. |
parallelism | 10 | Number of concurrent upload threads per table. |
| Region | 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 endpoints are only reachable from within the Treasure Data platform (for example, from a TD Workflow task) and cannot be accessed from the public internet.
The reactor_instance parameter routes the upload request but does not restrict which instances can read the data. All instances within the account share the same lookup catalog storage — any Lookup Catalog Attribute in any instance can read any table uploaded under the account, regardless of which reactor_instance value was used to upload it. If your account operates multiple RT 2.0 instances that must not share catalog data, do not rely on reactor_instance for isolation; use separate accounts or contact Treasure AI Support.
- In Data Workbench, create a new Workflow project (or use an existing one).
- Copy the
lookup-catalog-sync/directory from realtime-box/lookup-catalog-sync/ in the treasure-boxes repository and upload the files (lookup_catalog_sync.dig,scripts.py,queries/discover_tables.sql) into the project. - Edit
lookup_catalog_sync.digand setreactor_importer_endpoint,reactor_instance, andtd.databasein the_exportblock. Optionally setp_table_nameto restrict a run to a single table for testing. - Store your TD API key in the project's Secrets tab as
td.apikey. Use a key scoped to the minimum required permission (write access for the Bulk Load API) rather than a full-access key. - Set a schedule (daily is recommended) and run the workflow manually once to verify the initial full upload succeeds.
The discovery query automatically finds all base tables in cdp_lookup_catalog and excludes _wf_* internal tables.
Start with the default settings (batch_size: 1000, parallelism: 10) for the initial full upload. Increase gradually during incremental runs while monitoring for throttling errors. Larger tables or higher parallelism amplify the risk of throttling.
| Data type | Recommended schedule |
|---|---|
| Flash sale / time-sensitive pricing | Every 1–2 hours (0 */1 * * *) |
| Daily promotions / coupon tables | Every 6 hours (0 */6 * * *) |
| Product catalog (stable data) | Daily (0 2 * * *) |
| Symptom | Cause and fix |
|---|---|
| All records upload on every run | The digest table does not exist or has stale hashes. Drop _wf_{table_name}_digests and re-run to rebuild from scratch. |
400 error: "Expected string, received..." | A row has a NULL or empty primary key. Filter out or fix null-key rows in the source table before the next run. |
400 error: "Missing required header: x-lookup-table-name" | The x-lookup-table-name header is missing from the upload step. Verify the workflow template headers are configured correctly. |
Workflow fails with varchar = bigint type error | The source table column type was changed to an incompatible type (for example, string → int). Drop the digest table, fix the source data, and re-run. |
| Schema mismatch after changing the primary key column | Drop the _wf_{table_name}_digests table, then re-run. The workflow will re-upload all rows with the new key. Previously uploaded data under the old key is not automatically deleted. |
| Removed column still appears in responses | RT 2.0 stores data as uploaded — removing a column from the source table does not delete it from internal storage. A full re-upload is required to purge the residual data: drop the digest table (_wf_{table_name}_digests) and re-run the workflow. Until then, the deleted column's data remains accessible via the API. |
| Upload timeout | Reduce parallelism or batch_size, especially during the initial full upload of a large table. |
The workflow detects the primary key as the first non-time column in the table schema (by ordinal_position in information_schema.columns) — if the table's first column is named time, the second column is used as the primary key instead.
| Constraint | Detail |
|---|---|
| Column position | The key column must be the first non-time column in the table schema. |
| Column name | Must not be named time (reserved by Treasure AI platform). |
| Null values | Key column must not contain NULL or empty values. Upload fails with 400 if null keys are present. |
| Supported types | string. Other types are not supported for the primary key column. |
| Uniqueness | Data must be deduplicated. The workflow does not deduplicate rows. |