Erp Data Migration Patterns

ERP data migration is the process of transferring structured business data—customers, products, orders, inventory, financials—from legacy systems into a modern ERP platform. It matters when organizations replace old systems, consolidate operations, or adopt a new ERP suite. The migration is not a one-off task but a high-stakes engineering activity where data integrity, consistency, and performance are non-negotiable. This page details proven patterns for building reliable, scalable, and maintainable ERP data migrations, drawing from real-world practices across diverse environments.

Batch-First, Incremental-Second

Migrate large volumes of historical data in batches, then progressively update with deltas. Start with a full load of master data and transaction history, then run periodic incremental syncs.

Use a last_updated timestamp column on source tables, and maintain a migration_checkpoint table in the target ERP. Each batch migration starts by querying the source system for records updated after the last checkpoint. The checkpoint is updated after each batch completes.

-- Example checkpoint table structure
CREATE TABLE migration_checkpoint (
    entity_name VARCHAR(64) PRIMARY KEY,
    last_updated TIMESTAMP NOT NULL,
    batch_id VARCHAR(64),
    processed_rows INT,
    run_start TIMESTAMP,
    run_end TIMESTAMP
);

Before each batch, read the current checkpoint:

SELECT last_updated FROM migration_checkpoint WHERE entity_name = 'Customer';

Then migrate data:

SELECT *
FROM customer
WHERE updated_at > '2024-01-15T08:00:00Z'
ORDER BY updated_at
LIMIT 10000;

After writing to the ERP, update the checkpoint:

INSERT INTO migration_checkpoint (entity_name, last_updated, batch_id, processed_rows, run_start, run_end)
VALUES ('Customer', '2024-01-16T09:00:00Z', 'batch-20240116-01', 9842, '2024-01-16T08:00:00Z', '2024-01-16T09:00:00Z')
ON CONFLICT (entity_name)
DO UPDATE SET
    last_updated = EXCLUDED.last_updated,
    batch_id = EXCLUDED.batch_id,
    processed_rows = EXCLUDED.processed_rows,
    run_end = EXCLUDED.run_end;

Failure mode: The last_updated column is not indexed, causing full scans on large tables. Migration slows from minutes to hours. Sharp edge: Use updated_at as a float (e.g., TIMESTAMP WITH TIME ZONE) and ensure all systems use the same time zone; a one-hour difference between source and target causes missed records.

Idempotent Load with Duplicate Detection

Ensure that the same batch can be rerun without creating duplicates or overwriting data. Use a unique identifier (e.g., external_id or source_system_id) to detect and resolve duplicates.

Each record in the source has a source_system_id field. During migration, the ERP checks for existing records by this ID before inserting or updating.

-- Upsert logic using MERGE (SQL standard)
MERGE INTO customer AS target
USING (VALUES
    ('CUST-001', 'Acme Inc.', '2024-01-15T08:00:00Z', '2024-01-16T09:00:00Z'),
    ('CUST-002', 'Global Widgets', '2024-01-15T08:05:00Z', '2024-01-16T09:05:00Z')
) AS source (external_id, name, created_at, updated_at)
ON target.external_id = source.external_id
WHEN MATCHED THEN
    UPDATE SET
        name = source.name,
        updated_at = source.updated_at
WHEN NOT MATCHED THEN
    INSERT (external_id, name, created_at, updated_at)
    VALUES (source.external_id, source.name, source.created_at, source.updated_at);

Failure mode: The external_id is not enforced as a unique constraint. A duplicate external_id is silently inserted, leading to two identical customers with different internal IDs. Sharp edge: Set the external_id in the source system *before* the first migration batch. If the ID is generated during load, subsequent batches may create new records instead of updating existing ones.

Staging and Validation Pipeline

Migrate data into a staging environment before promoting it to production. Validate data through a sequence of checks: schema, business rules, referential integrity, and consistency.

Use a staging schema in the target ERP database. Migrate data to staging.customer, staging.invoice, etc. After the load, run validation queries:

-- Check for missing parent records
SELECT i.id, i.customer_id
FROM staging.invoice i
WHERE NOT EXISTS (
    SELECT 1 FROM staging.customer c
    WHERE c.external_id = i.customer_id
);

-- Check for unbalanced GL entries
SELECT account, SUM(debit) - SUM(credit) AS net_balance
FROM staging.gl_entry
GROUP BY account
HAVING ABS(SUM(debit) - SUM(credit)) > 0.01; -- Threshold: 1 cent

Use psql scripts to run these checks and generate a validation report:

#!/bin/bash
set -e

echo "Running validation checks..."
psql -d staging_erp -c "
SELECT 'Missing customer' as check, COUNT(*) as count
FROM staging.invoice i
WHERE NOT EXISTS (
    SELECT 1 FROM staging.customer c
    WHERE c.external_id = i.customer_id
)
UNION ALL
SELECT 'Unbalanced GL', COUNT(*)
FROM staging.gl_entry
GROUP BY account
HAVING ABS(SUM(debit) - SUM(credit)) > 0.01
"

Failure mode: The staging environment uses a different time zone than production. When validating date-based logic (e.g., fiscal periods), results differ between environments. Sharp edge: Store validation results in a validation_report table with run_id, check_name, status, details, and timestamp. Use jsonb for details to store error stacks, row samples, and SQL output.

Parallel, Chunked Loading

For large datasets, split the migration into chunks and process them in parallel. Use partition_id or batch_id to group records and distribute work.

Split customer data by region:

• region = 'NA' → chunk 1
• region = 'EMEA' → chunk 2
• region = 'APAC' → chunk 3

Each chunk is processed independently, often in separate worker processes or containers.

docker run --name customer-migrator-na -e CHUNK=NA -d customer-migration:latest
docker run --name customer-migrator-emea -e CHUNK=EMEA -d customer-migration:latest
docker run --name customer-migrator-apac -e CHUNK=APAC -d customer-migration:latest

Each worker reads the source data filtered by region, and writes to a staging schema:

SELECT *
FROM customer
WHERE region = 'NA'
ORDER BY updated_at
LIMIT 10000;

Use partition_id to track progress:

INSERT INTO migration_checkpoint (entity_name, last_updated, batch_id, partition_id, processed_rows)
VALUES ('Customer', '2024-01-16T09:00:00Z', 'batch-20240116-01', 'NA', 12456);

Failure mode: Workers write to the same staging table without coordination. Two chunks write to staging.customer simultaneously, causing interleaved data and inconsistent indexing. Sharp edge: Use partition_id as a prefix in file names and log paths: logs/customer/NA/20240116-01.log. This enables per-chunk debugging and rollbacks.

Incremental Delta Synchronization

After the initial batch load, maintain up-to-date data with real-time or near-real-time delta synchronization.

Use a change data capture (CDC) stream from the source system. For example, a change_log table records every insert, update, or delete:

CREATE TABLE change_log (
    id SERIAL PRIMARY KEY,
    entity_name VARCHAR(64) NOT NULL,
    operation VARCHAR(10) NOT NULL, -- 'INSERT', 'UPDATE', 'DELETE'
    entity_id VARCHAR(64) NOT NULL,
    payload JSONB NOT NULL,
    recorded_at TIMESTAMP NOT NULL
);

A background service polls the change_log table every 30 seconds:

import psycopg2
import json
import requests

conn = psycopg2.connect("dbname=erp user=loader password=secret")
cursor = conn.cursor()

last_id = 0
while True:
    cursor.execute("""
        SELECT id, entity_name, operation, entity_id, payload, recorded_at
        FROM change_log
        WHERE id > %s
        ORDER BY id
        LIMIT 100
    """, (last_id,))

    for row in cursor:
        id, entity_name, operation, entity_id, payload, recorded_at = row
        payload_dict = json.loads(payload)

        if operation == 'INSERT':
            requests.post("https://erp.api.com/v1/customer", json=payload_dict)

        elif operation == 'UPDATE':
            requests.put(f"https://erp.api.com/v1/customer/{entity_id}", json=payload_dict)

        elif operation == 'DELETE':
            requests.delete(f"https://erp.api.com/v1/customer/{entity_id}")

        last_id = id

    conn.commit()
    # Update a checkpoint table
    cursor.execute(
        "INSERT INTO migration_checkpoint (entity_name, last_updated, batch_id) VALUES (%s, %s, %s)",
        ('ChangeLog', recorded_at, f"delta-{last_id}")
    )
    conn.commit()

Failure mode: The change_log table is not indexed on id and recorded_at. Polling becomes slow as the log grows. Sharp edge: Use recorded_at as a cursor for incremental polling. If the service crashes, it resumes from the last recorded_at value instead of id, avoiding duplicated work.

Error Handling and Retry Strategy

Every migration step must handle failures gracefully. Errors are not logged and forgotten—they are tracked, retried, and escalated.

For each migration step, maintain an error_log table:

CREATE TABLE error_log (
    id SERIAL PRIMARY KEY,
    step_name VARCHAR(128) NOT NULL,
    entity_name VARCHAR(64),
    entity_id VARCHAR(64),
    error_code VARCHAR(32),
    message TEXT NOT NULL,
    raw_data JSONB,
    retry_count INT DEFAULT 0,
    last_attempt TIMESTAMP,
    created_at TIMESTAMP DEFAULT NOW(),
    resolved BOOLEAN DEFAULT FALSE
);

When a batch fails, log the error:

INSERT INTO error_log (step_name, entity_name, entity_id, error_code, message, raw_data, retry_count, last_attempt)
VALUES (
    'Customer Load',
    'Customer',
    'CUST-001',
    'ValidationFailed',
    'Customer name is null',
    '{"name": null, "external_id": "CUST-001"}',
    2,
    '2024-01-16T09:05:00Z'
);

Use a retry mechanism: each error is reprocessed every 15 minutes. After three failed attempts, escalate to a monitoring system.

Failure mode: A migration step logs an error but does not update the last_updated checkpoint. The next batch starts before the current one finishes, causing data loss or duplication. Sharp edge: Include the source system’s original row as raw_data in the error log. This enables quick debugging without requerying the source system.

Pre- and Post-Migration Scripts

Use scripts to prepare the environment before migration and verify it afterward.

Before migration:

• Truncate staging tables
• Create indexes on critical columns
• Set foreign key constraints
• Populate lookup tables

-- Pre-migration script
TRUNCATE staging.customer, staging.invoice, staging.inventory RESTART IDENTITY;
CREATE INDEX idx_customer_external_id ON staging.customer (external_id);
ALTER TABLE staging.customer ADD CONSTRAINT fk_customer_region FOREIGN KEY (region_id) REFERENCES staging.region(id);

After migration:

• Verify record counts
• Run reconciliation reports
• Update ERP metadata (e.g., last_migration_run)

-- Post-migration script
CREATE TABLE reconciliation_report (
    report_id SERIAL PRIMARY KEY,
    batch_id VARCHAR(64),
    source_count INT,
    target_count INT,
    diff INT,
    run_at TIMESTAMP,
    summary JSONB
);

INSERT INTO reconciliation_report (batch_id, source_count, target_count, diff, run_at)
SELECT
    'batch-20240116-01',
    (SELECT COUNT(*) FROM customer),
    (SELECT COUNT(*) FROM staging.customer),
    (SELECT COUNT(*) FROM customer) - (SELECT COUNT(*) FROM staging.customer),
    NOW();

Failure mode: The post-migration script runs on a different database than the one being migrated. The reconciliation_report table is populated in the wrong schema. Sharp edge: Store pre- and post-migration scripts in a migrations/scripts/ directory with pre-*.sql, post-*.sql naming. Use psql -f pre-01-customer.sql to run them reliably.

Summary

ERP data migration is not a single command or a one-off ETL. It is a series of interlocking patterns: batch-first, idempotent, staged, chunked, incremental, and resilient. Each pattern addresses a common failure mode: duplication, missing data, slow performance, uncoordinated workers, and undetected errors. Together, they form a robust foundation for any ERP data migration, scalable from small to enterprise-grade deployments.

This page was rewritten on 17 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.