Auto Loader — incremental file ingestion and schema evolution handled files landing in cloud storage. The other half of ingestion is message buses — Kafka, Azure Event Hubs, Kinesis — a live firehose of events rather than files. The exam objective names them explicitly, and they change how you ingest even though the medallion goal is unchanged.

The spine

Beat 1 — land it raw first: bronze is your replayable history

A message bus is transient — Kafka retains messages only for days. So the first move is not to transform on the way in.

Predict: an app writing to Kafka omitted a critical field. If you'd parsed and dropped "unneeded" columns at ingest, what have you lost?

…

The original data forever — the bus already aged it out. So the rule:

Anchor. Ingest all raw data and metadata from the bus into a bronze Delta table exactly as received (a streaming read → append). That bronze table is a permanent, replayable history of the source's state — if a downstream transform is wrong, or a field turns out to matter later, you replay from bronze instead of re-reading a bus that no longer has the data.

(spark.readStream.format("kafka")
   .option("kafka.bootstrap.servers", "host:9092")
   .option("subscribe", "orders")
   .option("startingOffsets", "latest")
   .load()                      # key, value (bytes), topic, partition, offset, timestamp
   .writeStream.option("checkpointLocation", "/chk/bronze")
   .toTable("bronze_orders"))

Keep key, value, and the metadata (topic, partition, offset, timestamp) — that's the audit trail. It's Structured Streaming (Structured Streaming & the state model), so it carries a checkpoint that tracks the consumed offsets.

Beat 2 — late data: the watermark, again

Bus events arrive out of order (network, retries, partitions).

Predict: a Kafka stream aggregates by event-time window, but records arrive after their window. What keeps the aggregation correct without holding state forever?

…

The watermark — the exact tool from Structured Streaming & the state model. withWatermark("event_time", "10 minutes") tells Spark how much lateness to tolerate before finalizing (and dropping) a window's state. "Kafka + late-arriving data + aggregation" → watermark. Same idea, message-bus framing.

Beat 3 — duplicates the bus can't dedupe: MERGE on a key

Message buses give at-least-once delivery — the same message can arrive twice, sometimes days apart.

Predict: duplicates can land a week apart. Can a watermark (which only holds a short window of state) dedupe those?

…

No — a watermark bounds state to a short window, so a duplicate a week later is outside it. For cross-history dedup you MERGE INTO on a unique key (WHEN NOT MATCHED THEN INSERT) against the target — the idempotent insert-only merge from Deduplication — distinct, keep-latest, and the streaming state trap. Bounded-window repeats → dropDuplicatesWithinWatermark; arbitrarily-far-apart repeats → insert-only MERGE.

Lock it. Bus ingest = stream-read → raw bronze (replayable) → watermark for late data → insert-only MERGE for far-apart duplicates.

The dials (skim now; return when a question needs one)

◆ The format matrix

The objective names many formats. Two ways to read them:

• Auto Loader (cloudFiles) — incremental file ingestion, supports JSON, CSV, Parquet, Avro, ORC, Text, and binaryFile (images/PDFs → raw bytes). The default for "new files keep arriving."
• spark.read.format(...) — batch read for any format, including XML (format("xml") with the rowTag option to pick the record element) and one-off ORC/Text loads.

Tell: "incrementally ingest files, schema evolution" → Auto Loader; "read this XML" → spark.read.format("xml").option("rowTag", …); "ingest images/PDFs" → binaryFile.

◆ Reading the bus — the key options

• Kafka: kafka.bootstrap.servers, subscribe (topic) / subscribePattern, startingOffsets (earliest/latest/explicit JSON). The value column is bytes — cast/parse it (from_json, etc.) after landing raw.
• Azure Event Hubs / AWS Kinesis — same Structured-Streaming shape, different format and connection options. All are unbounded streaming sources with checkpointed offsets.

◆ Append-only, batch and streaming

The bronze read above is append-only — new events append, nothing rewrites. The same Delta bronze table serves both: a streaming read for continuous ingest, and a batch read (spark.read) for reprocessing. That dual batch+streaming capability on one append-only Delta table is the "append-only pipeline for both" objective (and why bronze is where replay lives).

Takeaways (rebuild it from these)

1. Message buses (Kafka/Event Hubs/Kinesis) are transient streaming sources — land raw to bronze first (replayable history); keep key/value/offset metadata. It's Structured Streaming → checkpointed offsets.
2. Late data → watermark (same tool as Structured Streaming & the state model).
3. Duplicates far apart → insert-only MERGE on a unique key (watermark can't catch week-apart repeats).
4. Formats: Auto Loader = JSON/CSV/Parquet/Avro/ORC/Text/binaryFile (incremental files); spark.read.format("xml") + rowTag for XML; binaryFile for images/PDFs.
5. Bronze is append-only and serves both batch and streaming — the append-only-pipeline objective.

Before you move on — say these without scrolling up

1. Why land bus data raw to bronze before transforming?
2. Kafka aggregation with late-arriving records — what keeps it correct?
3. Duplicates a week apart — why won't a watermark catch them, and what does?
4. Read a batch of XML — which API and which option?

Next: cleaning and shaping what you've ingested → Section 3.