Home / Data Engineer Track / Streaming & Incremental Ingestion
DE Track · Module 03

Streaming & Incremental Ingestion

Structured Streaming, Auto Loader, and the incremental patterns behind every modern ingestion question.

The Structured Streaming mental model

Structured Streaming treats a stream as an unbounded table: new records are rows that keep getting appended, and your query is a normal DataFrame transformation that Spark re-evaluates incrementally as data arrives. You write the logic once; the engine figures out what changed since the last run and processes only that delta. Execution happens in micro-batches — small bounded jobs scheduled by a trigger — which is why a streaming query on Databricks behaves like a fast loop of batch jobs rather than record-at-a-time processing.

This is the single most useful framing for interviews, because it collapses "batch vs streaming" into one question: how often do you trigger the increment? Run it continuously and you have a low-latency stream. Run it once a day with Trigger.AvailableNow and you have an incremental batch job that never re-reads old data. Same code, same checkpoint, different schedule.

Sources and sinks on Databricks

On Databricks the dominant pattern is files or events in, Delta out, hop by hop through Bronze → Silver → Gold, with each hop a streaming read of the previous Delta table.

Checkpointing: the part you must get exactly right

Every streaming query owns a checkpoint location — a directory the engine uses to record its progress. Inside it stores:

Rules you should be able to recite cold:

Exactly-once semantics

Structured Streaming into Delta gives end-to-end exactly-once from two ingredients working together:

Break either ingredient and you degrade to at-least-once. The classic break: using forEachBatch to write to a non-transactional target (JDBC, REST API) without making the write idempotent yourself — e.g., keying a MERGE on a natural key, or tracking batchId in the target. Also note forEachBatch can re-execute a batch on failure, so any side effects inside it must tolerate replay.

Pitfall: "we'll just restart it from scratch." Deleting a checkpoint to "fix" a stuck stream is the streaming equivalent of rm -rf on your bookkeeping. Auto Loader's ingested-file ledger lives in that checkpoint — delete it and the next run re-ingests every file still in the source path into your Bronze table as duplicates. If you genuinely need a reset, do it deliberately: new checkpoint plus a truncated/recreated target, or a dedup-aware MERGE downstream. And never point two streams at one checkpoint to "share progress" — each query needs its own.

Triggers: how often the increment runs

# "Streaming as batch": incremental run that drains the backlog, then exits.
(spark.readStream
    .table("finance_bronze.gl_postings")
    .where("doc_status = 'POSTED'")
    .writeStream
    .option("checkpointLocation", chk_path + "/gl_silver")
    .trigger(availableNow=True)
    .toTable("finance_silver.gl_postings"))

Trigger.AvailableNow is the answer to more interview questions than any other trigger. "How would you make a nightly file load incremental without tracking filenames yourself?" "How do you backfill then switch to scheduled runs?" "How do you control cost on a low-volume stream?" — all three resolve to: write it as a stream, run it with AvailableNow on a job schedule, let the checkpoint do the bookkeeping.

Output modes

Watermarks, late data, and state

Stateful operations — windowed aggregations, stream-stream joins, dropDuplicates — must buffer state, and without a bound that state grows forever. A watermark is your declaration of how late data is allowed to be: "track the max event time seen, subtract 10 minutes; anything older is too late." The engine then knows when a window is finalized, can emit it in append mode, and can purge its state.

# Tolerate 10 minutes of lateness; count events per 5-minute window.
from pyspark.sql import functions as F

(events
    .withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "5 minutes"), "plant_id")
    .agg(F.count("*").alias("event_count"))
    .writeStream
    .outputMode("append")
    .option("checkpointLocation", chk_path + "/plant_counts")
    .toTable("ops_silver.plant_event_counts"))

Trade-off to articulate: a longer watermark catches more late stragglers but holds more state and delays results in append mode; a shorter watermark is cheap and fast but silently drops anything later than the threshold. There is no free lunch — pick based on how the source actually misbehaves.

Stateful joins, at a high level: stream-static joins (stream joined to a Delta dimension table) are stateless on the stream side and just work — the static side is re-read per batch, so dimension updates are picked up. Stream-stream joins buffer both sides in state and need watermarks plus a time-range join condition so the engine knows when buffered rows can be evicted. If you're asked to design one, lead with "define watermarks on both sides and a join window, or state grows unbounded."

"Your stream emits no results for windowed aggregations, or drops some late records. Walk me through why."

Strong outline: (1) In append mode a windowed aggregate emits only when the watermark passes the window end — if the source goes quiet, the watermark stops advancing (it's driven by observed event time, not wall-clock), so the last window can sit unemitted. (2) Records older than max(event_time) − watermark delay are dropped by design; check whether the producer's clock skew or batch upload pattern exceeds the delay. (3) Mitigations: lengthen the watermark, or land everything in Bronze in append mode without aggregation and aggregate in a later hop where you can reprocess. Bonus point: state cleanup is what the watermark buys you — without it, the job eventually dies of state bloat, so "just remove the watermark" is not a fix.

Auto Loader (cloudFiles): incremental file ingestion done for you

Auto Loader is a Structured Streaming source (format("cloudFiles")) that incrementally ingests new files from cloud storage. It solves the problem every hand-rolled batch loader eventually drowns in: which files have I already processed? Auto Loader keeps that ledger in the checkpoint (a scalable key-value store), so it handles millions of files without a control table, processes each file exactly once, and recovers cleanly from failures.

# Auto Loader: incremental JSON ingestion into Bronze with schema evolution.
(spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", chk_path + "/wire_bronze/schema")
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .load("abfss://landing@adlsacct.dfs.core.windows.net/webwire/")
    .select("*", F.col("_metadata.file_path").alias("source_file"),
            F.current_timestamp().alias("ingested_at"))
    .writeStream
    .option("checkpointLocation", chk_path + "/wire_bronze")
    .trigger(availableNow=True)
    .toTable("finance_bronze.webwire_events"))

What makes it more than a file watcher

Be precise about this in interviews: production ingestion at ADM, the Port Authority, and Maersk was batch and metadata-driven — ADF pipelines parameterized from control metadata, driving PySpark loads into Delta on ADLS Gen2. You built and operated exactly the file-tracking and watermarking machinery Auto Loader replaces, which is why you can explain its value concretely: every audit column, rerun-safety check, and control-table update your framework handled by hand is what cloudFiles gives you out of the box. Auto Loader and COPY INTO themselves are platform knowledge and lab practice for you, not production claims — and saying so cleanly builds trust before you go deep on the mechanics.

COPY INTO vs Auto Loader

DimensionCOPY INTOAuto Loader
InterfaceSQL command, batch semanticsStructured Streaming source (cloudFiles)
File trackingTracks loaded files internally; idempotent on rerunCheckpoint-backed ledger; exactly-once per file
Scale sweet spotThousands of files; degrades as directories grow largeMillions to billions of files; notification mode for high arrival rates
Schema evolutionLimited (explicit options, more manual)First-class: inference, evolution modes, _rescued_data
LatencyScheduled batch onlyContinuous, micro-batch, or AvailableNow — your choice
Operator profileSQL-first teams, simple one-off or low-volume loadsEngineering teams, ongoing ingestion pipelines
Current guidanceLegacy-leaning; fine for ad hoc loadsDatabricks' recommended default for file ingestion (and what Lakeflow Spark Declarative Pipelines use under the hood)

Decision rule of thumb: default to Auto Loader for any recurring file ingestion; reach for COPY INTO when a SQL-only user needs a quick idempotent load of a modest directory. As of mid-2026, Databricks docs steer new ingestion toward Auto Loader or managed Lakeflow Connect connectors, with COPY INTO positioned for simpler cases.

Backfill and reprocessing patterns

Batch vs streaming: the decision framework

FactorFavors batch / AvailableNowFavors continuous streaming
Latency requirementConsumers act hourly/daily (finance close, reporting)Decisions degrade in minutes (fraud, ops monitoring, inventory)
Source behaviorFiles/extracts arrive on a schedule; upstream is itself batchTrue event firehose (Kafka/Event Hubs), continuous arrivals
CostCluster runs minutes per day; serverless jobs spin up on demandAlways-on compute; justify it with the latency requirement
Complexity & opsSimple retries, easy reprocessing, junior-friendly runbooksWatermarks, state, checkpoint hygiene, 24/7 monitoring
Semantics neededPoint-in-time consistency at a cut-off (e.g., "as of close")Freshest-possible view, eventual completeness acceptable

The senior-engineer answer is rarely "stream everything." It's: write pipelines as incremental streaming code, then choose the trigger that matches the business latency requirement — which is usually AvailableNow on a schedule, upgraded to ProcessingTime only where someone can articulate the cost of staleness.

You have a real fit-gap story here. At ADM, part of your role is making architecture recommendations — explicitly including batch vs near-real-time trade-offs — to stakeholders for Finance R2R workloads. Sources like JDE, SAP, and HFM produce data on accounting cadences, and CFO-level consumers act on close-cycle and reporting timelines, so your recommendation weighed latency need against always-on compute cost and operational complexity, and landed on scheduled batch with tight orchestration (ADF + Lakeflow Jobs + Redwood RunMyJobs) rather than streaming for its own sake. Tell it that way: you didn't avoid streaming because you couldn't build it — you advised against paying for latency nobody would consume.

"Would you use Auto Loader or COPY INTO to ingest daily SAP extracts landing in ADLS, and why?"

Strong outline: (1) Default to Auto Loader: checkpoint-tracked exactly-once file ingestion, schema evolution with _rescued_data for the inevitable extract format drift, and it scales as file counts grow. (2) Run it with Trigger.AvailableNow on a Lakeflow Jobs schedule — batch economics, streaming bookkeeping. (3) Directory listing mode suffices at daily-extract volume; mention notification mode as the scale escape hatch. (4) COPY INTO is acceptable if the team is SQL-only and volume is small, but you'd still pick Auto Loader for an ongoing pipeline. (5) Close honestly: your production loaders were metadata-driven ADF + PySpark, so you can compare the hand-rolled approach to Auto Loader from direct experience of what the framework had to manage.

When asked anything about "incremental," name the bookkeeping explicitly. Weak answers say "we process only new data." Strong answers say where the progress is recorded: source offsets and the file ledger in the checkpoint, batch IDs in the Delta log for sink idempotency, and the schema history at the schema location. Interviewers probe exactly these seams.

Self-check before moving on

PreviousDelta Lake Deep Dive