How to Build a Data Migration Pipeline: ETL Patterns and Validation

This page describes how to build a data migration pipeline using standard ETL (Extract, Transform, Load) patterns and validation techniques. It applies when moving data between systems, especially when source and target schemas differ, or when you need to ensure data quality during transfer.

Understand Migration Scope and Requirements

Before writing code, define what data needs migration, its source and target locations, and how the process should handle errors. Use a manifest file to track which tables or datasets are being migrated.

Example manifest structure:

migrations:
  - source_table: users
    target_table: customer_dim
    primary_key: user_id
    columns:
      - name: user_id
        type: integer
      - name: email
        type: string
    validation:
      method: row_count
      expected: 100000

The manifest should include column mappings, primary keys, and any transformation rules. This approach allows you to version control your migration steps and validate them separately.

Extract Data From Source Systems

Use SQL SELECT statements to extract data from source systems. For large tables, implement pagination using LIMIT and OFFSET or cursor-based approaches.

Example extraction query:

SELECT user_id, email, created_at 
FROM users 
WHERE updated_at >= '2023-01-01' 
ORDER BY user_id 
LIMIT 1000 OFFSET 0;

If the source database supports it, use FOR UPDATE or LOCK IN SHARE MODE to prevent concurrent modifications during extraction.

Some databases require specific syntax for consistent reads:

-- PostgreSQL
BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;
SELECT * FROM users WHERE id > 1000;

-- MySQL
START TRANSACTION WITH CONSISTENT SNAPSHOT;
SELECT * FROM users WHERE id > 1000;

Transform Data During Migration

Apply transformations to match target schema requirements. Use standard SQL or scripting languages like Python or Bash for complex logic.

For example, converting a timestamp from one timezone:

UPDATE users 
SET created_at = CONVERT_TZ(created_at, 'UTC', 'America/New_York')
WHERE created_at IS NOT NULL;

Handle data type mismatches explicitly:

SELECT 
  CAST(id AS BIGINT) AS user_id,
  TRIM(email) AS email,
  DATE(created_at) AS created_date
FROM source_table;

When migrating data with nested structures, flatten or serialize them appropriately. Some systems support JSON types:

SELECT 
  id,
  JSON_EXTRACT(metadata, '$.name') AS name,
  JSON_EXTRACT(metadata, '$.preferences') AS preferences
FROM users;

Validate Migration Results

Implement validation checks to ensure data integrity. Common methods include row count comparison, checksums, and sample record verification.

Row count validation:

SELECT 
  (SELECT COUNT(*) FROM source.users) AS source_count,
  (SELECT COUNT(*) FROM target.customer_dim) AS target_count;

Checksum validation:

-- Generate checksum for source data
SELECT MD5(GROUP_CONCAT(CONCAT(id, email, created_at) ORDER BY id)) AS source_checksum
FROM users;

-- Compare against target
SELECT MD5(GROUP_CONCAT(CONCAT(user_id, email, created_date) ORDER BY user_id)) AS target_checksum
FROM customer_dim;

Validation can also include checking that all expected records are present:

SELECT COUNT(*) 
FROM source.users s
LEFT JOIN target.customer_dim t ON s.id = t.user_id
WHERE t.user_id IS NULL;

Handle Migration Failures

Migration failures can occur during any phase. Always implement retry logic and error logging.

For extraction failures, log the query and its result:

ERROR: Query failed with SQLSTATE 23505
Detail: duplicate key value violates unique constraint "users_pkey"
Query: INSERT INTO users (id, email) VALUES (12345, '[email protected]')

Use transactional approaches to roll back changes if validation fails:

BEGIN;
-- Migration steps here
COMMIT; -- or ROLLBACK;

Monitor Progress and Log Events

Track migration progress using a status table or log file. Record start time, completion time, number of rows processed, and any errors encountered.

A minimal status tracking table:

CREATE TABLE migration_log (
  id SERIAL PRIMARY KEY,
  migration_name VARCHAR(255),
  start_time TIMESTAMP,
  end_time TIMESTAMP,
  rows_processed INTEGER,
  status VARCHAR(50),
  error_message TEXT
);

Implement Data Quality Checks

Run data quality checks after loading. These include null value checks, format validation, and range constraints.

Example check for valid email format:

SELECT COUNT(*) 
FROM users 
WHERE email NOT REGEXP '^[^@]+@[^@]+\.[^@]+$';

Check for out-of-range values:

SELECT COUNT(*) 
FROM users 
WHERE age < 0 OR age > 150;

Use Incremental Migration Patterns

For large datasets, implement incremental migration to reduce load on source systems. Track the last processed record using a timestamp or ID.

Example incremental extraction:

SELECT * FROM users 
WHERE updated_at > '2023-06-01 00:00:00' 
AND updated_at < '2023-06-01 01:00:00';

Schedule and Automate Migration Runs

Schedule regular migration runs using cron jobs or orchestration tools. Ensure that each run has a unique identifier to track execution.

Example cron entry:

0 2 * * * /usr/local/bin/migrate_users.sh --batch-id 2023060102

Handle Data Conflicts and Duplicates

Use merge or upsert operations when the target system supports them. Ensure that primary key constraints are respected.

Example upsert pattern:

INSERT INTO customer_dim (user_id, email, created_date)
VALUES (12345, '[email protected]', '2023-06-01')
ON CONFLICT (user_id) 
DO UPDATE SET 
  email = EXCLUDED.email,
  created_date = EXCLUDED.created_date;

Validate After Migration

After completing the migration, perform final validation to ensure correctness.

Check for missing records:

SELECT COUNT(*) 
FROM source.users s
LEFT JOIN target.customer_dim t ON s.id = t.user_id
WHERE t.user_id IS NULL;

Compare data distributions:

SELECT 
  COUNT(*) as total,
  COUNT(CASE WHEN age < 18 THEN 1 END) as minors,
  COUNT(CASE WHEN age >= 18 AND age <= 65 THEN 1 END) as adults,
  COUNT(CASE WHEN age > 65 THEN 1 END) as seniors
FROM users;

Test Migration in Staging Environment

Before running migration in production, test with a subset of data. This includes testing all transformation rules and validation checks.

Run a small batch migration:

./migrate.sh --source-db staging_db --target-db prod_db --batch-size 1000

Handle Schema Changes

If source schema changes during migration, update your extraction logic accordingly. Maintain backward compatibility where possible.

Example handling of missing columns:

SELECT 
  COALESCE(old_column, new_default_value) AS column_name,
  other_columns
FROM source_table;

Manage Migration Rollbacks

Keep backup copies of target data before migration begins. If a migration fails or produces incorrect results, roll back to the previous state.

Example rollback procedure:

# Rename current table
mv customer_dim customer_dim_migrated

# Restore from backup
cp customer_dim_backup customer_dim

# Revert migration log
UPDATE migration_log SET status = 'rolled_back' WHERE migration_id = '20230601';

Monitor Resource Usage

Monitor CPU, memory, and I/O usage during migration runs. Long-running migrations may cause system performance issues.

Check system metrics during execution:

top -p $(pgrep -f migrate_script)

Document Migration Steps

Maintain documentation of all steps taken during migration. This includes configuration files, SQL scripts, and validation queries.

Example documentation entry:

Migration: Users to Customer Dimension
Date: 2023-06-01
Source: users table in staging_db
Target: customer_dim in prod_db
Validation: Row count check, MD5 checksum comparison

Clean Up Migration Artifacts

Remove temporary tables or files created during migration. Ensure that no orphaned data remains.

Example cleanup:

DROP TABLE IF EXISTS temp_user_import;
DELETE FROM migration_log WHERE status = 'completed';

This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.