Cloud Lab: Matillion CDC Pipeline

Overview

Matillion is a cloud-native ETL platform designed specifically for data warehouses like Snowflake, BigQuery, and Redshift. Unlike code-based CDC tools, Matillion provides a visual, drag-and-drop interface for building data pipelines with built-in CDC capabilities.

Key features:

• Visual pipeline designer with pre-built CDC components
• Native integrations with cloud data warehouses
• Automated CDC tracking using timestamp/hash columns or CDC logs
• No-code/low-code approach reduces development time
• Built-in orchestration, scheduling, and monitoring

CDC Approaches in Matillion:

1. Timestamp-based CDC: Track changes via updated_at column
2. Hash-based CDC: Compare row hashes to detect changes
3. Log-based CDC: Read the database transaction log. Matillion's Data Productivity Cloud now ships a native CDC pipeline (a Debezium-based CDC agent) for this; you no longer have to bolt on a separate tool. Older instance-based Matillion ETL relied on an external tool such as Debezium.

Prerequisites

• Matillion Instance: ETL for Snowflake/Redshift/BigQuery (14-day free trial available)
• Cloud Data Warehouse: Snowflake, Redshift, or BigQuery account
• Source Database: MySQL, PostgreSQL, SQL Server, or Oracle with CDC-ready schema
• Network Access: Source database accessible from Matillion (VPN/peering or public endpoint)
• Database Credentials: Read-only user on source, write access on target

Architecture

┌────────────────┐      ┌─────────────────┐      ┌──────────────┐
│   MySQL        │      │   Matillion      │      │  Snowflake   │
│   (Source)     │─────▶│   CDC Job        │─────▶│  (Target)    │
│   + updated_at │      │   (Orchestration)│      │              │
└────────────────┘      └─────────────────┘      └──────────────┘
                              │
                              │ Scheduling
                              ▼
                        ┌─────────────┐
                        │ Incremental │
                        │   Loads     │
                        │  (5 min)    │
                        └─────────────┘

Matillion orchestrates data movement between source and target. The current product, the Data Productivity Cloud, is a fully managed SaaS — you sign in to a hosted control plane and (optionally) run a lightweight agent inside your own network for private connectivity. The older, instance-based Matillion ETL ran as a self-hosted VM you provisioned yourself (an EC2 instance on AWS, or the equivalent on GCP/Azure); several steps below still reflect that legacy deployment and are flagged where they differ.

Lab Setup

Step 1: Get access to Matillion

Legacy self-hosted Matillion ETL — deploy from a cloud marketplace (Snowflake example):

1. Go to your cloud provider's marketplace (e.g. AWS Marketplace → Matillion ETL for Snowflake)
2. Select instance type (t3.medium minimum for testing, m5.large for production)
3. Configure VPC, subnets, and security groups (allow 443 for web UI)
4. Launch instance and note public IP/DNS
5. Access Matillion UI at https://<instance-ip>
6. Complete initial setup wizard (license activation, database connection)

Step 2: Configure Database Connections

In Matillion UI, go to Project → Manage Database Connections:

Source Connection (MySQL)

• Name: MySQL-Source
• Type: MySQL
• Hostname: mysql.example.com
• Port: 3306
• Database: production_db
• Username: matillion_reader
• Password: ••••••••
• SSL: Enabled (recommended)

Click Test to verify connectivity, then Save.

Target Connection (Snowflake)

• Name: Snowflake-Target
• URL: https://myaccount.snowflakecomputing.com
• Username: matillion_user
• Password: •••••••• (or use key-pair auth)
• Warehouse: MATILLION_WH
• Database: CDC_DB
• Schema: RAW_DATA
• Role: MATILLION_ROLE

Step 3: Prepare Source Table with CDC Column

Ensure your source table has a timestamp column for tracking changes:

-- MySQL: Add updated_at column if not present
ALTER TABLE customers 
ADD COLUMN updated_at TIMESTAMP 
DEFAULT CURRENT_TIMESTAMP 
ON UPDATE CURRENT_TIMESTAMP;

-- Create index for efficient CDC queries
CREATE INDEX idx_updated_at ON customers(updated_at);

-- Verify data
SELECT MAX(updated_at) FROM customers;

Step 4: Create Transformation Job (Initial Load)

In Matillion, create a new Transformation Job:

1. Click Project → New Job → Transformation
2. Name: CDC_Customers_Initial_Load
3. Drag components onto canvas:

Components:

• Table Input (MySQL-Source):
  • Connection: MySQL-Source
  • Table: customers
  • Sample Size: All Rows
• Rewrite Table (Snowflake-Target):
  • Target Table: RAW_DATA.CUSTOMERS_STAGING
  • Warehouse: MATILLION_WH
  • Drop Table: True (for initial load)

Connect Table Input → Rewrite Table and click Run.

Step 5: Create Orchestration Job (Incremental CDC)

Create an Orchestration Job for scheduled incremental loads:

1. Click Project → New Job → Orchestration
2. Name: CDC_Customers_Incremental

Orchestration Flow:

1. [Start]
    │
    ▼
2. [Set Variable: last_run_time]
    │  ← Query max(updated_at) from CUSTOMERS_STAGING
    │
    ▼
3. [Run Transformation: Extract_Incremental]
    │
    ▼
4. [Run Transformation: Merge_Into_Target]
    │
    ▼
5. [End]

Component Details:

1. Set Variable (Grid Variable: last_run_time)

• Type: SQL Query
• Connection: Snowflake-Target
• SQL:

SELECT COALESCE(MAX(updated_at), '1970-01-01')
FROM CDC_DB.RAW_DATA.CUSTOMERS_STAGING;

-- Re-read the last 5 minutes on every run; MERGE de-duplicates.
SELECT COALESCE(DATEADD(minute, -5, MAX(updated_at)), '1970-01-01')
FROM CDC_DB.RAW_DATA.CUSTOMERS_STAGING;

2. Transformation: Extract_Incremental

• Create new transformation job with these components:

• Table Input:
  • Connection: MySQL-Source
  • SQL Query:

  -- Use >= against the overlap watermark (already backed off by a
  -- safety margin). The idempotent MERGE absorbs the re-read rows.
  SELECT * FROM customers
  WHERE updated_at >= '${last_run_time}'
  ORDER BY updated_at;

  Order note: ORDER BY updated_at here only bounds the batch for readability — it is not a reliable event order. Wall-clock timestamps can tie or arrive out of commit order, so never treat the polled order as the true sequence of changes. Correctness comes from the primary-key MERGE applying the latest state, not from row order.

• Table Output:
  • Target: RAW_DATA.CUSTOMERS_INCREMENTAL
  • Mode: Append

3. Transformation: Merge_Into_Target

• Create transformation with SQL Script component:

MERGE INTO CDC_DB.RAW_DATA.CUSTOMERS_STAGING AS target
USING CDC_DB.RAW_DATA.CUSTOMERS_INCREMENTAL AS source
ON target.id = source.id
WHEN MATCHED THEN UPDATE SET
  target.name = source.name,
  target.email = source.email,
  target.updated_at = source.updated_at
WHEN NOT MATCHED THEN INSERT (id, name, email, updated_at)
VALUES (source.id, source.name, source.email, source.updated_at);

-- Clean up incremental table
TRUNCATE TABLE CDC_DB.RAW_DATA.CUSTOMERS_INCREMENTAL;

Step 6: Schedule the Orchestration Job

1. Open CDC_Customers_Incremental orchestration job
2. Click Schedule button (clock icon)
3. Enable: Yes
4. Frequency: Every 5 minutes (or as needed)
5. Click Save

Step 7: Add Monitoring Component

Add an Email Notification component for failures:

1. Drag Email component onto orchestration canvas
2. Configure SMTP settings in Project → Email Configuration
3. Set Email component to trigger On Failure (right-click component → Options)
4. Configure message with job variables:

Subject: CDC Job Failed: ${job_name}
Body:
Job Name: ${job_name}
Error: ${error_message}
Timestamp: ${timestamp}

Handling Deletes

Option A (preferred): move deletes out of CDC's blind spot

The cleanest fix is to stop hard-deleting at the source. Replace deletes with a soft-delete flag (is_deleted / deleted_at) that also bumps updated_at, so a deletion becomes an ordinary update your existing pipeline already captures:

-- Source: soft-delete instead of DELETE
UPDATE customers
SET is_deleted = TRUE, updated_at = CURRENT_TIMESTAMP
WHERE id = 42;

Then extend the target MERGE to honor the flag (mark, or physically remove, matched rows):

WHEN MATCHED AND source.is_deleted THEN DELETE
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED AND NOT source.is_deleted THEN INSERT ...;

Option B: detect deletes with a periodic anti-join snapshot

When you cannot change how the source deletes, reconcile against a full snapshot of the current source keys on a slower schedule (say nightly). Any target key that is absent from the source snapshot has been deleted. This is a full key scan, so run it far less often than the incremental poll:

-- 1. Land ALL current source primary keys into a snapshot table
--    (a full Table Input of `SELECT id FROM customers`).
-- 2. Anti-join: target keys not present in the source = deletions.
MERGE INTO CDC_DB.RAW_DATA.CUSTOMERS_STAGING AS target
USING (
  SELECT t.id
  FROM CDC_DB.RAW_DATA.CUSTOMERS_STAGING AS t
  LEFT JOIN CDC_DB.RAW_DATA.CUSTOMERS_KEY_SNAPSHOT AS s
    ON t.id = s.id
  WHERE s.id IS NULL          -- present in target, gone from source
) AS gone
ON target.id = gone.id
WHEN MATCHED THEN DELETE;      -- or: UPDATE SET is_deleted = TRUE

Prefer a soft delete (UPDATE SET is_deleted = TRUE) over a physical DELETE in the warehouse if downstream consumers need to see that a record was removed rather than have it silently vanish.

Advanced: Hash-Based CDC

For tables without updated_at column, use hash-based change detection:

Transformation Flow:

1. Table Input: Load full source table
2. Calculator: Generate hash of all columns
3. Filter: Compare with stored hashes in target
4. Table Output: Write only changed rows

Calculator Component (Generate Hash):

• Function: MD5 or SHA256
• Input: CONCAT(id, '|', name, '|', email, ...)
• Output Column: row_hash

Filter Component (Detect Changes):

• SQL:

SELECT source.*
FROM ${THIS} AS source
LEFT JOIN CDC_DB.RAW_DATA.CUSTOMERS_HASH_LOOKUP AS target
  ON source.id = target.id
WHERE target.row_hash IS NULL 
   OR source.row_hash != target.row_hash;

Note: Hash-based CDC requires full table scan each run, which can be expensive for large tables. Use timestamp-based CDC when possible.

Deletes: this filter compares source rows against stored hashes, so it detects inserts and updates but not deletes — a row removed from the source simply stops appearing in the scan and lingers in the target, exactly as with timestamp CDC. Because hash CDC already reads the full source each run, you get delete detection almost for free: anti-join the stored keys against the freshly scanned keys (the Handling Deletes pattern) and remove or flag the target keys that no longer appear.

Verification

1. Test Incremental Load

-- Insert new record in MySQL source
INSERT INTO customers (name, email) 
VALUES ('David Brown', 'david@example.com');

-- Update existing record
UPDATE customers 
SET email = 'bob.smith.updated@example.com' 
WHERE name = 'Bob Smith';

-- Wait for next scheduled run (or trigger manually in Matillion)

-- Verify in Snowflake target
SELECT * FROM CDC_DB.RAW_DATA.CUSTOMERS_STAGING 
WHERE name IN ('David Brown', 'Bob Smith');

2. Check Job History

In Matillion UI:

• Go to Task History (top menu)
• Filter by job name: CDC_Customers_Incremental
• Review execution times, row counts, and any errors

3. Monitor Performance

-- Check CDC lag
SELECT 
  MAX(updated_at) as latest_source_update,
  CURRENT_TIMESTAMP() as now,
  DATEDIFF(minute, MAX(updated_at), CURRENT_TIMESTAMP()) as lag_minutes
FROM CDC_DB.RAW_DATA.CUSTOMERS_STAGING;

-- Check ingestion rate
SELECT 
  DATE_TRUNC('hour', updated_at) as hour,
  COUNT(*) as records_loaded
FROM CDC_DB.RAW_DATA.CUSTOMERS_STAGING
WHERE updated_at > DATEADD(day, -1, CURRENT_TIMESTAMP())
GROUP BY DATE_TRUNC('hour', updated_at)
ORDER BY hour DESC;

Matillion-Specific Gotchas

-- Convert to UTC in source query
SELECT *, CONVERT_TZ(updated_at, 'America/New_York', 'UTC') as updated_at_utc
FROM customers;

Best Practices

1. Use Staging Tables

Load raw CDC data to staging first, then apply transformations:

RAW_DATA.CUSTOMERS_STAGING  (raw CDC data)
    ↓
CURATED.CUSTOMERS_CLEAN     (transformed, deduplicated)
    ↓
ANALYTICS.DIM_CUSTOMERS     (analytics-ready)

2. Implement Idempotency

Use MERGE instead of INSERT to handle duplicate runs:

MERGE INTO target USING source ON target.id = source.id
WHEN MATCHED THEN UPDATE ...
WHEN NOT MATCHED THEN INSERT ...;

3. Add Audit Columns

Track when records were loaded into target:

ALTER TABLE CUSTOMERS_STAGING ADD COLUMN matillion_load_timestamp TIMESTAMP;

-- In Table Output component, set:
-- Populate Columns: matillion_load_timestamp = CURRENT_TIMESTAMP()

4. Version Control Jobs

Matillion integrates with Git for version control:

• Go to Project → Git Configuration
• Connect to GitHub/GitLab repository
• Commit job changes with descriptive messages
• Use branches for testing new CDC patterns

5. Monitor Warehouse Credits

Matillion jobs consume data warehouse credits. Monitor usage:

-- Snowflake: Check credit usage by Matillion
SELECT 
  WAREHOUSE_NAME,
  SUM(CREDITS_USED) as total_credits,
  COUNT(*) as query_count
FROM SNOWFLAKE.ACCOUNT_USAGE.WAREHOUSE_METERING_HISTORY
WHERE START_TIME > DATEADD(day, -7, CURRENT_TIMESTAMP())
  AND WAREHOUSE_NAME = 'MATILLION_WH'
GROUP BY WAREHOUSE_NAME;

Matillion vs. Code-Based CDC

Common Issues & Solutions

Job fails with "Connection timeout"

• Check security group allows Matillion instance IP to database port
• Verify database endpoint is correct (hostname/IP)
• Test connection in Manage Database Connections → Test

"Variable not found" error in transformation

• Ensure variable is set as Grid Variable (global), not job-scoped
• Check variable name matches exactly (case-sensitive): ${last_run_time}

High latency in incremental loads

• Add index on updated_at column in source database
• Reduce frequency if loading < 100 rows per run (increase to 15 min)
• Use batch window: WHERE updated_at BETWEEN ... AND ... instead of > last_run_time

Duplicate records in target

• Verify MERGE key matches primary key of source table
• Check for NULL values in join columns (exclude or handle separately)
• Use QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY updated_at DESC) = 1 to deduplicate