Skip to main content

Data Migration Engine

Read every row of a legacy table, reshape it into a new schema, and write it to the target in batches. Rows that fail validation go to a rejects file with the reason, instead of into the target. Running the job again is safe: rows already migrated are skipped.

Components: sql_select → mapping → switch output, with sql_insert for valid rows and file for rejects.

Needs: network access from the node to both databases, and a connection string (DSN) for each. The runs used PostgreSQL 14 for both. This is a batch copy of what is in the table when the job starts; it does not follow later changes.

Complete job​

The mapping splits "Last, First" names, normalises email, parses MM/DD/YYYY dates, decodes one-letter status codes, and turns integer cents into a decimal string. An email without @ is marked for rejection, and an unknown status code throws; the output sends both kinds of row to the rejects file. The runs exercised the invalid-email case only.

data-migration.yaml
name: data-migration
type: pipeline
# One-shot job: stop after one failed execution instead of retrying.
restart_policy: never
selector:
  match_labels:
    pipeline_role: migration
config:
  input:
    sql_select:
      driver: postgres
      dsn: "postgres://USER:PASS@SRC_HOST:5432/SRC_DB"
      table: customers
      columns: [cust_no, full_name, email, signup, status_code,
        balance_cents]
      suffix: ORDER BY cust_no
  pipeline:
    processors:
      - mapping: |
          let parts = this.full_name.split(",")
          let email = this.email.trim().lowercase()
          let cents = this.balance_cents.number().int64()
          root.id = this.cust_no
          root.last_name = $parts.index(0).trim()
          root.first_name = $parts.index(1).trim()
          root.email = $email
          root.signed_up_on = this.signup.ts_strptime(
            "%m/%d/%Y"
          ).ts_format("2006-01-02", "UTC")
          root.status = match this.status_code {
            "A" => "active"
            "I" => "inactive"
            "S" => "suspended"
            _ => throw("unknown status code " + this.status_code)
          }
          # Signed cents: format the absolute value, then prefix the
          # sign, so -150 becomes -1.50 and -5 becomes -0.05.
          let sign = if $cents < 0 { "-" } else { "" }
          let abs = if $cents < 0 { 0 - $cents } else { $cents }
          root.balance = "%s%d.%02d".format(
            $sign, ($abs / 100).floor().int64(), ($abs % 100).int64()
          )
          root.reject_reason = if !$email.contains("@") {
            "invalid email"
          } else { null }
  output:
    switch:
      cases:
        - check: errored() || this.reject_reason != null
          output:
            file:
              path: "/var/tmp/expanso-migration/out/rejects.jsonl"
              codec: lines
            processors:
              - mapping: |
                  root = this
                  root.error = if errored() { error() } else { null }
        - output:
            sql_insert:
              driver: postgres
              dsn: "postgres://USER:PASS@DST_HOST:5432/DST_DB"
              table: customers
              columns: [id, first_name, last_name, email,
                signed_up_on, status, balance, migrated_from]
              args_mapping: |
                root = [
                  this.id, this.first_name, this.last_name,
                  this.email, this.signed_up_on, this.status,
                  this.balance,
                  "legacy.customers"
                ]
              suffix: ON CONFLICT (id) DO NOTHING
              max_in_flight: 4
              batching:
                count: 100
                period: 1s

The selector sends the job to a node labelled pipeline_role: migration; set that label on your node, or change the selector to a label it already carries. With no matching node, the job is stored and never runs.

Replace the two DSNs with your own. The runs used two local databases without TLS, with ?sslmode=disable on both DSNs; add it only in that situation.

The rejects path is on the node that runs the job; create its directory there first:

mkdir -p /var/tmp/expanso-migration/out

Try it on synthetic data​

The target table needs a primary key on id for ON CONFLICT (id) to work. This is the target schema the runs used:

target.sql
CREATE TABLE customers (
  id            integer PRIMARY KEY,
  first_name    text    NOT NULL,
  last_name     text    NOT NULL,
  email         text    NOT NULL UNIQUE,
  signed_up_on  date    NOT NULL,
  status        text    NOT NULL
    CHECK (status IN ('active','inactive','suspended')),
  balance       numeric(12,2) NOT NULL,
  migrated_from text    NOT NULL,
  migrated_at   timestamptz NOT NULL DEFAULT now()
);

And a synthetic source of 1,000 rows, 6 of them with an invalid email on purpose:

legacy.sql
-- Rows where cust_no % 143 = 0 (6 rows: 143..858) have an
-- invalid email on purpose.
CREATE TABLE customers (
  cust_no       integer PRIMARY KEY,
  full_name     text    NOT NULL,          -- "Last, First"
  email         text    NOT NULL,          -- mixed case, stray spaces
  signup        text    NOT NULL,          -- MM/DD/YYYY
  status_code   char(1) NOT NULL,          -- A / I / S
  balance_cents text    NOT NULL           -- integer cents as text
);
INSERT INTO customers
SELECT n,
       'Surname' || n || ', Given' || n,
       CASE WHEN n % 143 = 0 THEN ' user' || n || '.example.invalid '
            ELSE ' User' || n || '@Example.INVALID ' END,
       to_char(date '2019-01-01' + (n * 3), 'MM/DD/YYYY'),
       (ARRAY['A','I','S'])[1 + n % 3],
       ((n * 7919) % 500000)::text
FROM generate_series(1, 1000) AS n;

Load them into two separate databases, with SOURCE_DSN and TARGET_DSN set to their connection strings, then deploy:

psql "$TARGET_DSN" -v ON_ERROR_STOP=1 -f target.sql
psql "$SOURCE_DSN" -v ON_ERROR_STOP=1 -f legacy.sql
expanso-cli job validate data-migration.yaml --offline
expanso-cli job deploy data-migration.yaml

Check it​

Count the rows that landed, and the rejects on the node:

psql "$TARGET_DSN" -Atc "SELECT count(*) FROM customers"
wc -l < /var/tmp/expanso-migration/out/rejects.jsonl

With the synthetic data, expect 994 and 6. Run the job a second time and the target count stays at 994.

What the run proved​

  • 1,000 source rows produced 994 target rows and exactly 6 rejects, each carrying "reject_reason": "invalid email".
  • Every target row was identical to the same transform written independently in SQL.
  • A second run left the target unchanged at 994 rows.
  • The job was deployed through Expanso Cloud to a labelled node running expanso-edge v2.1.21, against two local PostgreSQL 14.23 databases. Those Cloud runs used the job without restart_policy.
  • The job with restart_policy: never, before the balance fix below, was run again on a local-mode node running v2.1.21: 994 target rows and 6 invalid-email rejects, the target identical to the independent SQL transform, and a second run completed with the target still at 994 rows. Each run completed after 1 execution.
  • Negative balances, local-mode node only. The balance formatter was corrected after the runs above, which had only positive balances: the earlier one turned -150 cents into -1.-5, which PostgreSQL rejects. The published job was run on a local-mode node against 9 rows from -99999 to 12345 cents. All 9 were written as exact decimals (for example -1.50, -0.05, -999.99, 0.00, 123.45), identical to an independent SQL transform, after 1 execution. A second run left the table unchanged; the rejects file gets its reject lines again on every run.

Limits​

  • Only PostgreSQL to PostgreSQL was run. sql_select and sql_insert accept other drivers (see their component pages); those were not part of the run.
  • The rejects file is on the node. Replace that file output to send rejects somewhere else.
  • ON CONFLICT (id) DO NOTHING skips rows that already exist; it does not update them. Change the suffix if you need an upsert.
  • The published job, with restart_policy: never, was proved on a local-mode node only; the Cloud runs used the earlier job without it. With never, a failed run stops after one execution instead of being re-run. See Proving your own run.