Skip to content
Last updated

Lookup Catalog Sync Workflow

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.

Workflow files on GitHub

The latest workflow files are available in the treasure-data/treasure-boxes repository under realtime-box/lookup-catalog-sync/:

  • lookup_catalog_sync.dig — main workflow
  • scripts.py — generates type-aware JSON payload SQL per table
  • queries/discover_tables.sql — discovers eligible tables in cdp_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.

How It Works

The workflow follows the same per-table sync pipeline for every discovered table:

Yes

No

Discover tables

For each table

Extract changed records
via hash comparison

Changes exist?

Upload to RT 2.0
Internal Storage

Skip — no update

Update digest table

Cleanup temp tables

Yes

No

Discover tables

For each table

Extract changed records
via hash comparison

Changes exist?

Upload to RT 2.0
Internal Storage

Skip — no update

Update digest table

Cleanup temp tables

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.

Configuration Parameters

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.databasecdp_lookup_catalogSource 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_size1000Number of records per upload request to the Bulk Load API.
parallelism10Number of concurrent upload threads per table.

Region-Specific Endpoints

Region reactor_importer_endpoint
UShttps://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
Tokyohttps://bulk-storage-importer-api-production-aws-tokyo-ap-northeast-1.internal.treasuredata.com
Internal endpoint access

.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.

No instance-level data isolation

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.

Setting Up the Workflow

  1. In Data Workbench, create a new Workflow project (or use an existing one).
  2. 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.
  3. Edit lookup_catalog_sync.dig and set reactor_importer_endpoint, reactor_instance, and td.database in the _export block. Optionally set p_table_name to restrict a run to a single table for testing.
  4. 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.
  5. 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.

Parallelism guidance

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.

Scheduling Recommendations

Data type Recommended schedule
Flash sale / time-sensitive pricingEvery 1–2 hours (0 */1 * * *)
Daily promotions / coupon tablesEvery 6 hours (0 */6 * * *)
Product catalog (stable data)Daily (0 2 * * *)

Troubleshooting

Symptom Cause and fix
All records upload on every runThe 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 errorThe source table column type was changed to an incompatible type (for example, stringint). Drop the digest table, fix the source data, and re-run.
Schema mismatch after changing the primary key columnDrop 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 responsesRT 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 timeoutReduce parallelism or batch_size, especially during the initial full upload of a large table.

Key Column Constraints

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 positionThe key column must be the first non-time column in the table schema.
Column nameMust not be named time (reserved by Treasure AI platform).
Null valuesKey column must not contain NULL or empty values. Upload fails with 400 if null keys are present.
Supported typesstring. Other types are not supported for the primary key column.
UniquenessData must be deduplicated. The workflow does not deduplicate rows.