CDC, REST/JSON & Streaming Ingestion

The one-liner

Which ingestion path? — click a source

▸ match the source to the right mechanism

1 · Read from Kafka / Event Hubs

subscribe (exact) / subscribePattern (regex) / assign (partitions) select topics. startingOffsets = latest or earliest; maxOffsetsPerTrigger caps batch size so a backlog doesn't overwhelm you.

PySpark

df = (spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", "host:9092")
        .option("subscribe", "events")            # or subscribePattern / assign
        .option("startingOffsets", "latest")      # "earliest" to backfill
        .option("maxOffsetsPerTrigger", 500000)   # cap per micro-batch
        .load())

SQL — read_kafka streaming TVF

CREATE OR REFRESH STREAMING TABLE cat.sch.bronze_raw AS
SELECT * FROM STREAM read_kafka(bootstrapServers => 'host:9092', subscribe => 'events');

2 · Kafka rows are binary — deserialize the value

▸ toggle to see why you must parse value

Deserialize + keep metadata

from pyspark.sql.functions import col, from_json
schema = "user_id INT, event STRING, ts TIMESTAMP"
parsed = (df.select(
            from_json(col("value").cast("string"), schema).alias("d"),  # deserialize
            col("topic"), col("partition"), col("offset"), col("timestamp"))
          .select("d.*", "topic", "partition", "offset", "timestamp"))   # keep offsets for replay

3 · Write to Delta — exactly-once via checkpoint

The checkpoint gives exactly-once + restart safety (one checkpoint per stream). Event Hubs is consumed through the same Kafka reader (Kafka protocol). Track lag with avgOffsetsBehindLatest.

(parsed.writeStream
   .option("checkpointLocation", "/Volumes/cat/sch/vol/_ck_kafka")  # unique per stream
   .trigger(availableNow=True)        # batch-style catch-up; omit for always-on
   .toTable("cat.sch.bronze_events"))

4 · Database CDC & REST/JSON

• Database CDC: mirror an OLTP DB via a Lakeflow Connect database connector (gateway + staging + CDC, 4.3); apply changes into silver dimensions with AUTO CDC / APPLY CHANGES in a Declarative Pipeline (5.4). Don't hand-roll change tracking.
• REST/JSON: handle auth/pagination/rate-limits, land raw JSON to a Volume, then ingest with Auto Loader.

# After landing raw JSON to a Volume, parse incrementally with Auto Loader
raw = (spark.readStream.format("cloudFiles")
         .option("cloudFiles.format", "json")
         .option("cloudFiles.schemaLocation", "/Volumes/cat/sch/vol/_api_schema")
         .load("/Volumes/cat/sch/vol/api_landing"))

Pick the path per source

Why it matters

• Real businesses have databases and event streams, not just files — choosing the right path per source is core DE judgment.
• "How do you get real-time events / database changes into the lakehouse?" → Kafka/Event Hubs streaming + CDC via Lakeflow Connect.

Uses, edge cases & limitations

• Uses: real-time events (Kafka/Event Hubs), database mirroring (CDC), API/JSON pulls into bronze.
• Edge cases: Kafka value is binary — must deserialize; startingOffsets=earliest backfills everything (cap with maxOffsetsPerTrigger); REST needs auth/pagination/rate-limit handling.
• Limitation: message-bus streaming is for events, not big file dumps (use Auto Loader); hand-built REST lacks managed-connector monitoring — prefer Lakeflow Connect where available.

Common gotchas

• ❌ Not deserializing Kafka value (binary) → garbage columns.
• ❌ Accidentally using earliest offsets → massive unintended backfill.
• ❌ Sharing one checkpoint across streams → corrupted state.
• ❌ Hand-coding DB ingestion when a Lakeflow Connect CDC connector exists.
• ❌ Dropping the Kafka offset/partition metadata — keep it in bronze for replay.