What you will be able to do
- Configure Structured Streaming reads from Kafka, Kinesis and Google Pub/Sub with their required options
- Deserialize the binary payload each message bus returns into typed columns
- Use Trigger.AvailableNow for incremental batch ingestion from a message bus into a Delta table
- Avoid the operational mistakes specific to each connector: Kinesis stream identifiers, Pub/Sub duplicates and roles
1.Reading from Kafka
On Databricks, every message bus is read through Structured Streaming with a connector-specific format keyword. For Kafka, that keyword is kafka. Every Kafka read, batch or streaming, needs kafka.bootstrap.servers, a comma-separated list of host:port pairs. It also needs exactly one way of choosing which topics to read.
df = (spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "<server:ip>")
.option("subscribe", "<topic>")
.option("startingOffsets", "latest")
.load()
)| Option | Value | Use |
|---|---|---|
| subscribe | A comma-separated list of topics. | The topic list to subscribe to. |
| subscribePattern | Java regex string. | The pattern used to subscribe to topic(s). |
| assign | JSON string {"topicA":[0,1],"topic":[2,4]}. | Specific topicPartitions to consume. |
Each row also carries topic, partition, offset, timestamp and timestampType. SQL users can stream from Kafka with the read_kafka table-valued function in Databricks Runtime 13.3 LTS and above. Streaming SQL only works inside Lakeflow pipelines or with streaming tables in Databricks SQL. Batch reads are also supported: use spark.read with startingOffsets and endingOffsets.
Checkpoint 1 of 7· Check yourself
Where can you run a streaming SQL query that uses the read_kafka function?
The read_kafka function can stream only inside Lakeflow pipelines or in Databricks SQL streaming tables. Python and Scala use readStream with the kafka format.
“Streaming with SQL is supported only in Lakeflow pipelines or with streaming tables in Databricks SQL.”Source: docs.databricks.com
Sources1
2.From Kafka bytes to a Delta table
Because the payload is bytes, a Kafka ingestion pipeline always includes a parsing step. In Databricks' reference workflow, the key and value are cast to strings, parsed with from_json against a fixed schema, and then flattened. For Avro or Protobuf payloads, use from_avro or from_protobuf instead, optionally with a schema registry. The example below also authenticates with a Unity Catalog service credential and no inline secrets.
parsed_df = (spark.readStream
.format("kafka")
.options(**kafka_options)
.load()
.select(
from_json(col("key").cast("string"), key_schema).alias("key"),
from_json(col("value").cast("string"), value_schema).alias("value")
)
.select("key.*", "value.*")
)Many pipelines don't need a stream running all the time. For incremental batch loading, Databricks recommends Kafka with Trigger.AvailableNow. The query processes what is available, writes it to Delta with a checkpoint, and stops. Run on a schedule, it ingests incrementally without a cluster running nonstop.
Checkpoint 2 of 7· Fill the gap
Complete the write so the job processes everything currently available in Kafka and then stops.
query = (parsed_df.writeStream
.format("delta")
.option("checkpointLocation", "/path/to/checkpoint")
.trigger( ? =True)
.toTable("catalog.schema.events_table")
)availableNow=True gives incremental batch processing: the query consumes what is available, writes to the Delta table using the checkpoint, and terminates.
Source: docs.databricks.comSources1
3.Reading from Amazon Kinesis
Kinesis works the same way with format kinesis, and the payload is again opaque bytes. To deserialize the data column, cast it to a string. Kinesis can authenticate with a Unity Catalog connection, a service credential, an instance profile or access keys. Trigger.AvailableNow works with Kinesis in Databricks Runtime 13.3 LTS and above. You can also reshard by adding shards without restarting the stream.
| Column | Type | Description |
|---|---|---|
| partitionKey | string | The partition key that identifies which shard the record is assigned to. |
| data | binary | The record's data blob, as opaque binary. |
| stream | string | The name or ARN of the Kinesis stream the record was read from. |
| shardId | string | The ID of the shard the record was read from. |
| sequenceNumber | string | The unique identifier of the record within its shard. |
| approximateArrivalTimestamp | timestamp | The approximate time the record was inserted into the stream. |
The most testable Kinesis rule is how you identify the stream. You must set either streamName or streamARN, never both. streamARN needs Databricks Runtime 16.1 or above. Once a query is running, you cannot switch from one to the other: restarting with the other option can duplicate or lose records. In Databricks Runtime 19 and above, you can add a stream to the list and restart. By default, removing a stream fails the restart with KINESIS_SOURCE_STREAMS_REMOVED_ON_RESTART, so records from that stream are never silently skipped. Also note that each shard supports five GetRecords calls per second across all polling readers. With N polling queries on the same stream, set minFetchPeriod to at least 200ms × N.
Checkpoint 3 of 7· Check yourself
A production Kinesis query uses streamName. The team wants to identify the stream by streamARN instead. What is the supported approach?
You cannot switch between streamName and streamARN on an active query. Doing so can duplicate or lose data, so the switch needs a new query with a new checkpoint.
“To switch from streamName to streamARN, start a new streaming query with a fresh checkpoint directory.”Source: docs.databricks.com
Sources2
4.Subscribing to Google Pub/Sub
Pub/Sub uses format pubsub, or read_pubsub in a SQL streaming table. You supply subscriptionId, topicId and projectId, plus four credential options: clientEmail, clientId, privateKey and privateKeyId. Databricks recommends storing those keys as secrets, as in the SQL example below.
CREATE OR REFRESH STREAMING TABLE pubsub_raw
AS SELECT * FROM STREAM read_pubsub(
subscriptionId => 'mysub',
projectId => 'myproject',
topicId => 'mytopic',
clientEmail => secret('pubsub-scope', 'clientEmail'),
clientId => secret('pubsub-scope', 'clientId'),
privateKeyId => secret('pubsub-scope', 'privateKeyId'),
privateKey => secret('pubsub-scope', 'privateKey')
);The credentials need roles/pubsub.viewer (or roles/viewer) and roles/pubsub.subscriber. Add roles/pubsub.editor only if the connector should create missing subscriptions or delete them when the stream stops. The message body arrives in payload as raw bytes, next to messageId, a JSON string of attributes, and publishTimestampInMillis.
The connector gives exactly-once processing for the rows it receives from the subscriber. That is not the same as receiving no duplicates. Pub/Sub itself may publish a message twice or deliver rows out of order, and handling that is your job. The progress metric numDuplicatesSinceStreamStart shows how many duplicates the stream has seen. With Trigger.AvailableNow, the connector records a start timestamp and processes rows published before it. Speculative execution (spark.speculation) is not supported.
Checkpoint 4 of 7· Check yourself
A silver table fed from Pub/Sub occasionally shows the same messageId twice. What explains this?
Exactly-once applies to rows received from the subscriber. Pub/Sub may still deliver duplicates or out-of-order rows, and the documentation says you must write code to handle them.
“Pub/Sub might publish duplicate rows, or rows might arrive to the subscriber out of order.”Source: docs.databricks.com
Sources3
5.Same pattern, different payload columns
The three connectors share a design, and once you see it the exam questions become recognisable. Each has a format keyword and a SQL table-valued function for streaming tables. Each supports Trigger.AvailableNow for incremental batch runs. Each returns the message body as bytes plus some bus-specific metadata. In every case the decoding step is yours: cast to string, from_json, from_avro or from_protobuf, depending on how the producer encoded the data. The main difference is which column holds the body.
Checkpoint 5 of 7· Match them up
Match each source to the column that holds its raw message body
Tap a term, then the definition that fits it.
All three return the body as bytes in a differently named column, so every pipeline has to deserialize it explicitly.
“Use DataFrame operations (such as cast("string") or from_avro) to explicitly deserialize the keys and values.”Source: docs.databricks.com
Checkpoint 6 of 7· Exam question
A bronze-layer Auto Loader stream is defined without an explicit schema, so `cloudFiles.schemaEvolutionMode` defaults to `addNewColumns`. A new field starts appearing in the upstream JSON payload. What happens to the running stream?
Correct answer: A — The stream throws an `UnknownFieldException` and stops; once the orchestration layer restarts the job, the merged schema is picked up and the new column populates from that point onward.
- A. Correct. With `addNewColumns`, Auto Loader intentionally fails the micro-batch with an `UnknownFieldException` so the orchestrator restarts the stream with the evolved schema, after which new records populate the added column normally.
- B. Incorrect. Schema changes under `addNewColumns` are not applied mid-batch without a restart; the stream must stop and relaunch with the updated schema definition before the new column is usable.
- C. Incorrect. Silently dropping new fields describes the `none` evolution mode, not the `addNewColumns` default, which is designed to surface and incorporate new columns rather than discard them.
- D. Incorrect. Routing unexpected fields into `_rescued_data` without evolving the schema describes the `rescue` mode, which is a different, explicitly opted-into behavior from the `addNewColumns` default.
Checkpoint 7 of 7· Exam question
A data engineer is building a Structured Streaming job that reads order events directly from a Kafka topic into a bronze Delta table: ```python df = (spark.readStream .format("kafka") .option("kafka.bootstrap.servers", bootstrap) .option("subscribe", "orders") .option("startingOffsets", "latest") .load()) ``` The job must survive restarts without reprocessing or silently skipping messages, and must not crash the first time offsets fall outside the topic's retention window. Which changes should the engineer make? (Select all that apply.)(Select 3)
Correct answers: A, B, C — Set `startingOffsets` to `earliest` for the very first run so the stream begins from the oldest retained offset instead of only picking up events produced after the stream starts.; Attach a durable `checkpointLocation` on the write side so Kafka partition offsets and processing progress persist across restarts instead of being lost.; Set `failOnDataLoss` to `false` so the stream logs a warning and continues instead of aborting when expected offsets have already aged out of the topic.
- A. Correct. Using `latest` on the very first run would skip every event already sitting in the topic; starting from `earliest` on that initial run ensures the backlog is captured before ongoing incremental processing takes over.
- B. Correct. A checkpoint location is what lets Structured Streaming resume from the exact Kafka offsets it last committed, which is required to avoid reprocessing or skipping messages across restarts.
- C. Correct. When retention has already expired requested offsets, the stream fails by default; explicitly tolerating data loss lets the job continue from the earliest available offset instead of crashing outright.
- D. Incorrect. `assign` requires an explicit JSON map of specific topic partitions to consume; it does not mean subscribing to every partition of every topic, and dropping `subscribe` entirely would leave the topic selection undefined.
- E. Incorrect. Coalescing to a single partition after the read does not manage offset checkpointing at all, and it removes the parallelism Kafka partitions provide without addressing restart correctness.
- F. Incorrect. Relying on `startingOffsets=latest` on every restart, rather than only the first run, means the checkpoint is being ignored and any events produced while the stream was down would be permanently skipped.
Exam traps
Each one states something that sounds right. Open it to see what is actually true.
1.If the producer wrote JSON text, the Kafka source returns value as a string you can query directly.Why is that wrong?
Kafka keys and values always arrive as binary byte arrays. You must cast or decode them, for example with cast("string") then from_json, or with from_avro.
Covered in Reading from Kafka
2.Because the Pub/Sub connector is exactly-once, the ingested table never contains duplicate messages.Why is that wrong?
Exactly-once covers rows received from the subscriber. Pub/Sub may still publish duplicates and deliver out of order, so deduplication and ordering are your responsibility.
Covered in Subscribing to Google Pub/Sub
Sources
Every claim above is drawn from one of these pages, quoted as it was written on the date shown.
- 1.
“Streaming with SQL is supported only in Lakeflow pipelines or with streaming tables in Databricks SQL.”
↩︎ Reading from Kafka“For both batch and streaming queries, you must set the bootstrap servers for the Kafka source with the following option:”
↩︎ Reading from Kafka“For incremental batch loading, Databricks recommends using Kafka with Trigger.AvailableNow.”
↩︎ From Kafka bytes to a Delta table“For other formats like Avro or Protobuf, use from_avro or from_protobuf.”
↩︎ From Kafka bytes to a Delta table“The key and the value are always deserialized as byte arrays with the ByteArrayDeserializer.”
↩︎ Exam trap 1“The key and the value are always deserialized as byte arrays with the ByteArrayDeserializer.”
↩︎ Prediction“Use DataFrame operations (such as cast("string") or from_avro) to explicitly deserialize the keys and values.”
↩︎ Checkpoint - 2.
“To deserialize the data in the data column, cast the field to a string.”
↩︎ Reading from Amazon Kinesis“For all Databricks Runtime versions, you must specify either streamName or streamARN, but not both.”
↩︎ Reading from Amazon Kinesis“Kinesis supports authentication with a Unity Catalog connection, a service credential, or alternate methods such as an instance profile or access keys.”
↩︎ Reading from Amazon Kinesis“Each Kinesis shard supports five GetRecords calls per second across polling readers.”
↩︎ Reading from Amazon Kinesis“In Databricks Runtime 13.3 LTS and above, you can use Trigger.AvailableNow with Kinesis.”
↩︎ Same pattern, different payload columns“To switch from streamName to streamARN, start a new streaming query with a fresh checkpoint directory.”
↩︎ Checkpoint - 3.
“This connector has exactly-once processing semantics for rows from the subscriber.”
↩︎ Subscribing to Google Pub/Sub“Databricks recommends that you use secrets when using keys.”
↩︎ Subscribing to Google Pub/Sub“Pub/Sub does not support speculative execution with spark.speculation.”
↩︎ Subscribing to Google Pub/Sub“You can use Trigger.AvailableNow to consume available rows from the Pub/Sub sources as an incremental batch.”
↩︎ Same pattern, different payload columns“You must write code to handle duplicate and out-of-order rows.”
↩︎ Exam trap 2“Pub/Sub might publish duplicate rows, or rows might arrive to the subscriber out of order.”
↩︎ Checkpoint