Small-File Ingestion and At-Least-Once Merge Pipeline
A logistics company lands 40 million JSON order events per day in S3 as roughly 250,000 files of 40-300 KB, arriving continuously with up to 90 minutes of late arrivals. Snowpipe with auto-ingest loads them into RAW_EVENTS; a stream on RAW_EVENTS feeds a task DAG that MERGEs into SILVER_ORDERS every 5 minutes. Three problems: duplicate ORDER_ID rows appear in SILVER after Snowpipe retried on a transient error; Snowpipe and S3 GET costs are about 3x the original estimate; and analysts querying SILVER see data up to two hours old during peak. The team also occasionally sees a MERGE run start before the previous one finished, and they are considering replacing the whole chain with dynamic tables.
1. Where do the duplicate ORDER_IDs actually come from, and how do you make the pipeline idempotent?
Read the worked answer
Start by rejecting the assumption that Snowpipe is the culprit. Snowpipe tracks loaded files by path and checksum for 14 days and will not reload an identical file, so a plain retry of the same object does not duplicate rows. The realistic sources are:
- The file is not identical. An upstream process rewrites or renames a file, or re-sends it with a new name. Snowflake sees a new file - correctly - and loads it.
- The source re-sends the same ORDER_ID in a different file. At-least-once delivery upstream becomes at-least-once in the table.
- The MERGE itself is not idempotent. Two overlapping executions, or a MERGE that always UPDATEs on match without a watermark, can double-count.
Confirm which using COPY_HISTORY (was the same path loaded twice?) and the stream offset history.
The durable fix is to stop relying on source-side guarantees and make the merge deterministic:
- Deduplicate the source inside the MERGE with
QUALIFY ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY ingest_ts DESC) = 1. - Match on ORDER_ID and guard the update:
WHEN MATCHED AND src.ingest_ts > tgt.ingest_ts THEN UPDATE. - Force single-flight execution: one root task, children chained by dependencies,
ALLOW_OVERLAPPING_EXECUTION = FALSE. - Add a post-load assertion that counts duplicate ORDER_IDs and fails loudly, so the next regression is caught in minutes, not by an analyst.
2. Snowpipe and S3 costs are 3x estimate. How do you attack the small-file problem?
Read the worked answer
250,000 files of 40-300 KB is roughly a gigabyte-scale payload split into a quarter of a million units. Per-file overhead - metadata handling, queueing, serverless bookkeeping, S3 GET requests - dominates, and no amount of warehouse tuning touches it.
Decision framework driven by the 90-minute late-arrival tolerance:
- If 90 minutes of latency is genuinely acceptable, buffer and compact upstream. Have the producer or a stream processor roll events into ~50-250 MB files on a 5-10 minute boundary and land those. Two orders of magnitude fewer files cuts both Snowpipe and S3 costs immediately and gives better compression.
- If latency must be seconds, files are the wrong transport. Snowpipe streaming or the Kafka connector ingests rows directly, eliminating the file-per-event tax entirely, with lower cost and offset-based delivery semantics.
- At 40M events/day, do the arithmetic on both designs rather than assuming; the crossover point is usually driven by whether the producer can batch at all.
Also instrument before you change: PIPE_USAGE_HISTORY and COPY_HISTORY show credits per pipe and file counts; serverless task billing shows up separately from warehouse credits, and a 5-minute DAG of 1,400 tasks has its own cost that consolidating runs would remove. Do not fix the ingestion and leave a DAG that wakes up 288 times a day to move a few thousand rows.
3. How do you address the two-hour freshness lag and the overlapping MERGE, and is replacing the chain with dynamic tables sensible?
Read the worked answer
Overlapping runs first, because it explains the lag. If MERGE executions overlap, they contend on the same target and the later run waits or duplicates. Enforce a single root task with dependent children and ALLOW_OVERLAPPING_EXECUTION = FALSE, then examine TASK_HISTORY for SCHEDULED versus EXECUTED timestamps to see where the hours go: a 5-minute schedule that takes 40 minutes per run produces an eight-hour backlog regardless of scheduling. Common culprits are a MERGE reading a large unconsolidated stream, a queued warehouse, or a long dependency chain where each child waits on the slowest sibling.
Stream staleness. A stream advances only when the consuming DML commits. If a task remains broken or disabled past the stream's retention window the stream goes stale and must be recreated - check the stream's STALE_AFTER and alert on it, because a stale stream is silent until someone notices the dashboard stopped moving.
Dynamic tables. They are a good fit for the shape of this problem: declare TARGET_LAG and let Snowflake manage incremental refresh, watermarks, and scheduling, replacing hand-written merge logic and its overlap bugs. The trade-offs are real, though: a dynamic table cannot consume a stream, some constructs and non-deterministic functions are restricted, and you give up fine-grained control over exactly when refreshes run and which warehouse serves them. You also inherit refresh cost that is less obvious to attribute.
A pragmatic path: convert SILVER to a dynamic table with TARGET_LAG = '1 minute' if the transformation is expressible in the supported subset, keep RAW_EVENTS as the immutable landing zone, and retire the streams and DAG. If the merge logic depends on constructs the feature does not support, keep the task-based design and fix single-flight execution and warehouse placement instead - do not force the migration.