RIO Ingestion Service - CRM Data Sync Pipeline
Scheduled, incremental pipeline that extracts 29 tables from Actian CRM365 over JDBC, transforms and vaults PII, and fans the result out to S3 (bronze archive), ClickHouse (analytics, via S3Queue) and RDS PostgreSQL (operational upsert). Publishes an EventBridge audit event at every stage.
RIO Ingestion Service - CRM Data Sync Pipeline
What This Pipeline Does
This flow documents how CRM data is continuously copied into RIO. The CRM (Actian CRM365) is the system of record, and the RIO application cannot query it directly. So on a schedule, this pipeline pulls whatever changed since last time and fans it out to three purpose-built stores:
- S3 (bronze layer) — A permanent archive of every table. Think of it as a backup that can be used to rebuild the other stores if needed.
- ClickHouse (gold layer) — The analytics database that powers dashboards and the Cube semantic layer.
- RDS PostgreSQL - The transactional store the RIO application reads.
Each destination keeps track of how far it has synced. If one destination fails, the others keep working. The next run will automatically retry only what the failed destination missed.
How This Flow Starts
This pipeline is not triggered by events. It is started by a timer (AWS EventBridge Scheduler).
| Environment | How Often |
|---|---|
| Production | Every hour |
| Non-production | Every 6 hours |
There is also a second way the pipeline can start: when someone uploads a product master file to S3. In that case, the pipeline runs in a special “Product Master” mode (see Run Modes below).
Step-by-Step Walk-Through
| Step | What Happens | Key Detail |
|---|---|---|
| 1 | Timer triggers the pipeline | Runs every hour (prod) or every 6 hours (non-prod) |
| 2 | Service starts up and checks watermarks | Connects to all required services and checks how far each table was synced last time |
| 3 | CRMSyncStarted event is published | Announces the start of the run with a unique run ID that links all events together |
| 4 | Changed rows are extracted from the CRM | Only rows that changed since the last sync are read, in batches of 2,000 |
| 5 | Data is transformed and PII is hashed | Business rules are applied, dates and booleans are standardized, and personal data is hashed into a DynamoDB lookup table |
| 6 | Data is saved to S3 (bronze layer) | All 29 tables are archived here first. This is always the first write |
| 7 | Data is loaded into ClickHouse (gold layer) | 16 tables are written to ClickHouse for analytics. Duplicates are removed before writing. The pipeline waits up to 90 seconds to confirm the load |
| 8 | Data is loaded into RDS PostgreSQL | 8 source tables are loaded into 5 Postgres tables. New rows are inserted, existing rows are updated |
| 9 | ETLBatchCompleted event is published | One event per table. If a table failed, an ETLBatchFailed event is sent instead, and the pipeline moves on |
| 10 | HierarchyUpdatedByIngestion event is published | Only for the system users table, and only when a manager change is detected |
| 11 | CRMSyncCompleted event is published | Sent after all tables are done. If any table failed, CRMSyncFailed is sent instead |
Steps 4 through 10 repeat for each of the 29 tables, one at a time.
Where Each Table Lands
The pipeline processes 29 CRM tables. All 29 are archived to S3 (bronze). On top of that, 16 go to ClickHouse and 8 go to RDS. 5 tables go to S3 only.
- S3 (bronze archive) — All 29 tables. Permanent archive and rebuild source.
- ClickHouse (analytics) — 16 tables covering accounts, campaigns, contracts, products, territories, leads, opportunities, quotes, sales orders, entitlements, and BPF history. These power dashboards and the Cube semantic layer.
- RDS PostgreSQL (application) — 8 source tables loaded into 5 target tables: person, tenant_role, opportunity, account_team, and deal_team. Some source tables write to more than one target, which is why 8 sources produce 5 targets.
- S3 only — 5 tables (customer addresses, beta program participants, pipeline stages, software programs, and software program products). Archived for reference but not loaded into any database.
Run Modes
The pipeline can run in three modes:
| Mode | What It Does |
|---|---|
| Incremental (default) | Only syncs rows that changed since the last run. This is the normal mode used by the hourly schedule |
| Initial Load | Syncs all rows from scratch, without checking what was already synced. Used for first-time setup or full reloads |
| Product Master | Completely different from the other two. Skips the normal 29-table sync entirely. Instead, it reads an uploaded product master file from S3, compares it to what is already in ClickHouse, and only loads the changes. No CRM connection is used and no audit events are published |