Kafka to Databricks: Three Ways to Stream Into Delta Lake

Read Kafka on Databricks with the kafka format in Structured Streaming, or with read_kafka inside a Lakeflow pipeline, and write to a Delta table with a checkpoint. Use a streaming table when you want Databricks to manage orchestration, checkpoints and recovery. Use a Structured Streaming job when you need custom code. Auto Loader reads files, not topics, so it only fits when something else already lands the topic in object storage.

A Structured Streaming job that loads new records and stops:

(spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<bootstrap-server>:9092")
  .option("subscribe", "<topic-name>")
  .option("databricks.serviceCredential", "<service-credential-name>")
  .load()
  .selectExpr("CAST(key AS STRING) AS key", "CAST(value AS STRING) AS value", "topic", "partition", "offset", "timestamp")
  .writeStream
  .option("checkpointLocation", "/path/to/checkpoint")
  .trigger(availableNow=True)
  .toTable("catalog.schema.events_raw"))

The same ingestion as a streaming table:

CREATE OR REFRESH STREAMING TABLE catalog.schema.events_raw AS
SELECT * FROM STREAM read_kafka(
  bootstrapServers => '<bootstrap-server>:9092',
  subscribe => '<topic-name>',
  serviceCredential => '<service-credential-name>'
);

• kafka.bootstrap.servers plus one of subscribe, subscribePattern or assign is the minimum.
• startingOffsets defaults to latest for streams and applies only to a new query. A resumed query always reads from its checkpoint.
• failOnDataLoss defaults to true, so the query fails when offsets have expired from Kafka retention. Set it to false only if skipping data is acceptable.
• maxOffsetsPerTrigger caps each micro-batch, spread proportionally across partitions. minOffsetsPerTrigger with maxTriggerDelay (15 minutes by default) waits until a batch is worth running.

Key and value arrive as binary, so cast them or decode them with from_avro. The Kafka source does not commit offsets back to Kafka, so a lag dashboard built on consumer groups shows nothing for these streams. Read avgOffsetsBehindLatest and maxOffsetsBehindLatest from the query progress instead.

Databricks recommends streaming tables for most ingestion, and says streaming tables with continuous execution and enhanced autoscaling are the most efficient route for low-latency loading from message buses. They live in Lakeflow Spark Declarative Pipelines, the product formerly called Delta Live Tables.

• Streaming table with read_kafka when the work is ingestion and light transformation.
• Structured Streaming job when you need custom code, foreachBatch or several sinks.
• Auto Loader when a connector already writes the topic to object storage as files. It discovers files incrementally and cannot subscribe to a topic.

The checkpoint records which offsets each micro-batch read and which batches committed to the sink, and that is what gives exactly-once processing into a Delta table. Four rules prevent most outages:

• One checkpoint location per query. Never share one.
• Changing the subscribed topics needs a new checkpoint. Changing rate limits or the trigger interval does not.
• foreachBatch gives at-least-once, so make the batch logic idempotent.
• Configure the job to restart on failure. Databricks recommends the continuous trigger on jobs compute, which restarts a failed run with exponential backoff.

Exactly-once inside Spark does not change what Kafka retention does. If offsets expire before the stream catches up, the query fails or skips data. The Kafka side is covered in Kafka exactly-once semantics.

For Avro, from_avro takes a registry subject and address, so nobody maintains the schema by hand:

from_avro(
  data = col("value"),
  jsonFormatSchema = None,
  options = schema_registry_options,
  subject = "t-value",
  schemaRegistryAddress = "https://<registry-endpoint>"
)

For a Confluent registry, pass credentials in the options map: confluent.schema.registry.basic.auth.credentials.source set to USER_INFO, and confluent.schema.registry.basic.auth.user.info set to key:secret. By default from_avro fails on an incompatible schema change (FAILFAST). Set avroSchemaEvolutionMode to restart on Databricks Runtime 14.2 and above and the query throws when it sees the change, so a restart-on-failure job picks it up deliberately instead of at random.

Spark creates one partition per Kafka topic-partition unless you set minPartitions, so read parallelism starts at your partition count and cores beyond it sit idle. minPartitions and maxRecordsPerPartition split large Kafka partitions across more tasks, which helps with skew and catch-up, but minPartitions reinitialises the Kafka consumers on every trigger and can hurt performance over SSL. A hot partition still leaves one task slower than the rest, so fix the key before adding cores.

Size the cluster to the partition count at peak, with headroom. The tradeoffs are in Kafka partition strategy and Kafka consumer lag.

• Amazon MSK with IAM. On Databricks Runtime 16.1 and above, create a Unity Catalog service credential and set databricks.serviceCredential. Databricks recommends it for shared clusters and serverless compute. Do not combine it with kafka.sasl.* or kafka.security.protocol options. Instance profiles use AWS_MSK_IAM with the shaded IAM login module instead.
• SASL/PLAIN or SCRAM. Set kafka.security.protocol to SASL_SSL and use Databricks' shaded class names, such as kafkashaded.org.apache.kafka.common.security.scram.ScramLoginModule. Unshaded names fail with RESTRICTED_STREAMING_OPTION_PERMISSION_ENFORCED.
• Confluent Cloud. The API key and secret go in as the SASL/PLAIN username and password. Keep them in a secret scope, not in notebook code.

When connections time out, check the network before the credentials. The broker security group must admit the Databricks compute, typically on port 9098 for IAM, with peering or PrivateLink in place.

• Use Trigger.AvailableNow for incremental loads. Databricks recommends it for incremental batch loading from Kafka, and serverless notebooks and jobs support only it and Trigger.Once. For continuous streaming on serverless, use a pipeline in continuous mode.
• Run streams on jobs compute, not all-purpose compute, and do not enable autoscaling on Structured Streaming jobs.
• A processingTime trigger of 0 gives the lowest latency and raises cloud storage API costs. Ask whether the consumer needs seconds or minutes.
• Use minOffsetsPerTrigger to avoid a stream of tiny micro-batches, and liquid clustering (CLUSTER BY) to keep query performance as the table grows.

Several failure modes above sit on the Kafka side: expired offsets, a hot partition, an authentication change. If the cluster is the problem, that is Kafka support, or managed Kafka services if you would rather not run it. If the Databricks side needs design work, see Databricks consulting and Databricks implementation.

• Databricks: Connect to Apache Kafka
• Databricks: Kafka authentication
• Databricks: Read and write streaming Avro data
• Databricks: Spark API options reference
• Databricks: Structured Streaming checkpoints
• Databricks: Production considerations for Structured Streaming
• Databricks: Load data in pipelines
• Databricks: What happened to Delta Live Tables
• Apache Spark: Structured Streaming + Kafka Integration Guide
• Confluent: Connect self-managed Kafka clients to Confluent Cloud