How to build reproducible ingestion pipelines that do not flake out under load. You kicked off a backfill, the loader retried, and now the fact table has duplicates - analytics are off and you are diffing CSVs at 2 a.m. At billion-row scale, ingestion must be fast (parallel I/O, bulk paths), reproducible (reruns do not corrupt data), idempotent (retries do not create duplicates), and observable (you can prove what loaded and why).

This guide shows a production-grade pattern using COPY, staging tables, and idempotent upserts (MERGE) - with Snowflake-flavored SQL and Python, but the ideas map to BigQuery, Redshift, and Postgres.

Architecture at a Glance

Flow: Object Store → Stage → Raw/Staging → Dedup → Target

1. Land files in object storage (S3/GCS/Azure).
2. COPY into a raw staging table (append-only).
3. Normalize and deduplicate into a clean staging table.
4. MERGE into target tables with a deterministic upsert policy.
5. Audit: track file lineage, checksums, row counts, and load windows.

Why this works

• Bulk path: COPY uses massively parallel readers.
• Reproducible: raw staging is immutable; you can re-derive everything.
• Idempotent: MERGE enforces one logical row per business key.
• Auditable: file and row-level metadata survives the journey.

COPY vs INSERT vs Streaming - When to Use What

Ingestion mode	Best for	Pros	Cons
COPY (bulk files)	Backfills, daily/hourly batches, billions of rows	Parallel, cheap, predictable	Needs file prep + staging
INSERT (row-by-row)	Small trickle loads, admin tasks	Simple to code	Slow, expensive at scale
Streaming (CDC/queues)	Near-real-time	Low latency, continuous	Complexity, exactly-once is hard

Rule of thumb: use COPY for big volumes; add CDC/streams when latency truly matters. Even then, land micro-batches and COPY them.

Concrete Example (Snowflake)

1) Create stage and file format

CREATE STAGE mystage
  URL='s3://my-bucket/sales/'
  CREDENTIALS=(AWS_KEY_ID=$KEY AWS_SECRET_KEY=$SECRET);

CREATE OR REPLACE FILE FORMAT my_csv_fmt
  TYPE = CSV FIELD_OPTIONALLY_ENCLOSED_BY='"' SKIP_HEADER=1
  NULL_IF=('','NULL');

2) Raw staging table (append-only)

Keep source fidelity plus metadata. Use a stable business key and ingestion metadata.

CREATE OR REPLACE TABLE raw_sales (
  order_id        STRING,
  customer_id     STRING,
  order_ts        TIMESTAMP_NTZ,
  amount          NUMBER(18,2),
  src_file_name   STRING,
  src_row_num     NUMBER,
  load_ts         TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
  _sha1_row       STRING
);

3) COPY with validation and lineage

Validate first, then load. Record file names to prevent accidental re-COPY.

-- Dry run
COPY INTO raw_sales
  FROM @mystage
  FILE_FORMAT = (FORMAT_NAME = my_csv_fmt)
  VALIDATION_MODE = RETURN_ERRORS;

-- Actual load with columns aligned
COPY INTO raw_sales(order_id, customer_id, order_ts, amount, src_file_name, src_row_num, _sha1_row)
FROM (
  SELECT
    $1, $2, TO_TIMESTAMP_NTZ($3), $4,
    METADATA$FILENAME, METADATA$FILE_ROW_NUMBER,
    SHA1(TO_VARCHAR($1)||'|'||$2||'|'||$3||'|'||$4)
  FROM @mystage (FILE_FORMAT => 'my_csv_fmt')
);

Pro tip: maintain a loaded_files control table populated from INFORMATION_SCHEMA.LOAD_HISTORY so reruns can skip already consumed files.

4) Clean and de-duplicate into a staging table

Keep only the latest occurrence of each business key per file window.

CREATE OR REPLACE TEMP TABLE stage_sales AS
SELECT * EXCLUDE (rn)
FROM (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY order_id
      ORDER BY order_ts DESC, load_ts DESC
    ) AS rn
  FROM raw_sales
)
WHERE rn = 1;

5) Idempotent upsert into target

Use MERGE keyed by business key and update only when the data truly changed (checksum).

CREATE OR REPLACE TABLE sales AS
SELECT * FROM stage_sales WHERE 1=0;

MERGE INTO sales t
USING stage_sales s
  ON t.order_id = s.order_id
WHEN MATCHED AND t._sha1_row <> s._sha1_row THEN
  UPDATE SET
    customer_id = s.customer_id,
    order_ts    = s.order_ts,
    amount      = s.amount,
    src_file_name = s.src_file_name,
    src_row_num   = s.src_row_num,
    _sha1_row     = s._sha1_row,
    load_ts       = s.load_ts
WHEN NOT MATCHED THEN
  INSERT (order_id, customer_id, order_ts, amount, src_file_name, src_row_num, _sha1_row, load_ts)
  VALUES (s.order_id, s.customer_id, s.order_ts, s.amount, s.src_file_name, s.src_row_num, s._sha1_row, s.load_ts);

Why this is idempotent:

• Re-running COPY + stage_sales yields the same winning row per order_id.
• MERGE updates only when the checksum differs, so duplicate retries are harmless.

Python Orchestration Skeleton (Airflow/Dagster/Prefect)

Keep tasks deterministic and easy to re-run.

from pathlib import Path
import hashlib
import snowflake.connector

def sha1_file(p: Path) -> str:
    return hashlib.sha1(p.read_bytes()).hexdigest()

def load_files(files):
    con = snowflake.connector.connect(...)
    cur = con.cursor()
    try:
        for f in files:
            # idempotency guard: skip already loaded files
            cur.execute("SELECT 1 FROM control.loaded_files WHERE file_name=%s", (f.name,))
            if cur.fetchone():
                continue

            cur.execute(f"PUT file://{f} @mystage AUTO_COMPRESS=TRUE OVERWRITE=FALSE")
            cur.execute("""
                COPY INTO raw_sales(order_id, customer_id, order_ts, amount, src_file_name, src_row_num, _sha1_row)
                FROM (SELECT $1,$2,TO_TIMESTAMP_NTZ($3),$4,METADATA$FILENAME,METADATA$FILE_ROW_NUMBER,
                             SHA1(TO_VARCHAR($1)||'|'||$2||'|'||$3||'|'||$4))
                FILE_FORMAT=(FORMAT_NAME='my_csv_fmt')
            """)
            cur.execute("INSERT INTO control.loaded_files(file_name, file_sha1) VALUES(%s,%s)",
                        (f.name, sha1_file(f)))
        con.commit()
    finally:
        cur.close(); con.close()

Best Practices at Scale

File strategy

• Target 100-250 MB compressed file size for the COPY sweet spot.
• Avoid millions of tiny files; consolidate before load.
• Use partitioned prefixes: s3://bucket/sales/dt=2025-11-20/part-0001.gz.

Schema and keys

• Define a stable business key (order_id) early.
• Add a checksum column to detect meaningful change.
• Track source lineage: filename, row number, produced_at, producer_version.

Idempotency

• Maintain a control table for loaded_files(file_name, sha1, first_seen_ts).
• Build pure transformations from raw → stage so reruns are deterministic.
• Use MERGE with a change predicate (t._sha1_row <> s._sha1_row).

Reproducibility

• Version your file format, SQL, and code.
• Pin library versions; store a manifest of files loaded for each run.
• Treat raw staging as append-only; never mutate it.

Performance

• Increase warehouse/slot size during COPY+MERGE windows, then scale down.
• Use clustering/partitioning on the target by query predicates (e.g., order_ts).
• Prefer bulk MERGE over per-row upserts.

Observability

• Capture row counts: staged in, deduped, merged-inserted, merged-updated, rejected.
• Sample records for schema drift and null explosions.
• Alert on COPY errors, merge anomalies, and unexpected volume shifts.

Safety and recovery

• Use VALIDATION_MODE before big loads.
• Keep a quarantine table for rejects; never DROP raw until checks pass.
• Make reruns explicit by window (e.g., dt BETWEEN '2025-11-01' AND '2025-11-07').

Common Pitfalls

• Tiny files → compact upstream or in a pre-job.
• Slow UPSERTs → checksum compare plus bulk MERGE, not per-row loops.
• Duplicate keys post-merge → add a unique constraint/quality check on order_id.
• Retry storms → make tasks idempotent and retry-safe; never append without guards.
• Schema drift surprises → ingest to VARIANT or raw strings, cast in stage, not in target.
• Forgotten lineage → add src_file_name, src_row_num, producer_version everywhere.

Takeaways

If you can rerun yesterday without fear, you have built a real pipeline. The key moves:

• Land files → COPY into append-only raw.
• Normalize/dedup → stage with a deterministic winner per key.
• Load targets via MERGE with a change checksum.
• Track file lineage and row counts to prove correctness.
• Make every step idempotent and observable.

Wrap this into an Airflow/Dagster job with clear run windows and control tables, add CDC later if the business truly needs low latency, and document the contract: one logical row per order_id, last-write wins by order_ts.