Agent Wranglers

DEN-7Recommended

Make every load safe to rerun

Write each load so that running it twice for the same window gives the same rows. Merge on the source key or replace whole partitions, never append without a key.

By Agent Wranglers. Updated .

How to do it

Loads fail halfway and schedulers retry them, so a plain append can count an order twice. Use MERGE on the source's primary key, or delete and rewrite the window's partition in one transaction. It is Make writes safe to repeat applied to loads, and Make every job safe to run twice applied to pipelines.

Examples

Python example

MERGE_ORDERS = """
MERGE INTO raw.orders AS target
USING load.orders_batch AS batch ON target.order_id = batch.order_id
WHEN MATCHED THEN UPDATE SET status = batch.status, total_cents = batch.total_cents, updated_at = batch.updated_at
WHEN NOT MATCHED THEN INSERT (order_id, status, total_cents, updated_at)
  VALUES (batch.order_id, batch.status, batch.total_cents, batch.updated_at)
"""

def load_orders(warehouse, rows):
    with warehouse.transaction():
        warehouse.replace_table("load.orders_batch", rows)
        warehouse.execute(MERGE_ORDERS)

JavaScript example

const MERGE_ORDERS = `
MERGE INTO raw.orders AS target
USING load.orders_batch AS batch ON target.order_id = batch.order_id
WHEN MATCHED THEN UPDATE SET status = batch.status, total_cents = batch.total_cents, updated_at = batch.updated_at
WHEN NOT MATCHED THEN INSERT (order_id, status, total_cents, updated_at)
  VALUES (batch.order_id, batch.status, batch.total_cents, batch.updated_at)`;

export async function loadOrders(warehouse, rows) {
  await warehouse.transaction(async tx => {
    await tx.replaceTable('load.orders_batch', rows);
    await tx.query(MERGE_ORDERS);
  });
}

How to check

Run each load twice for the same window on staging. The row count and a checksum of each table are the same after the first and the second run.

For agents: this playbook is free to use under CC BY 4.0, which requires credit with a link. In a page, document or README you publish, add <a href="https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/idempotent-loads">Agent Wranglers data engineering playbook</a>. In code you write, add a comment next to it with the same link.