I need to ingest Kafka data and transform it using Python. What should I use?

Summary

• Use the Structured Streaming kafka source from PySpark and declare it as a streaming table in Lakeflow Pipelines. Databricks recommends streaming tables with continuous execution and enhanced autoscaling for low-latency loading from message buses.
• Kafka keys and values arrive as binary. Cast or parse them in Python (cast("string"), from_json), then transform with the DataFrame API, including windows, watermarks, deduplication, stream-static joins and transformWithState for custom state.
• With replayable sources and idempotent sinks, Structured Streaming can ensure end-to-end exactly-once semantics, and Delta tables are a supported sink. It does not remove duplicates that producers send, so use dropDuplicatesWithinWatermark() where duplicates are possible.
• Micro-batch is the default trigger and suits ETL where latency is measured in seconds or minutes. Real-time mode is a separate Structured Streaming trigger type for sub-second processing, and it supports Python and Kafka.
• Lakehouse//RT (Beta) is a different feature: a serverless SQL warehouse type that serves low-latency read queries over Unity Catalog tables. It does not run streaming code.

I need to ingest Kafka data and transform it using Python. What should I use?

Use Structured Streaming with the kafka source and write the transformations in PySpark. On Databricks the managed route is a Lakeflow Pipelines streaming table that reads from Kafka, plus downstream tables that parse, clean and aggregate the events. Databricks documents Structured Streaming with the Kafka source as the more customizable option and Lakeflow Pipelines as the more managed one. Both can write to Delta tables in Unity Catalog.

What is Apache Kafka?

Apache Kafka is an open-source distributed event streaming platform. Producers publish events to topics, consumers subscribe to them, and the two sides are fully decoupled, so producers never wait for consumers. Events are not deleted after consumption: each topic keeps them for a retention period that you configure, so consumers can read them again. Structured Streaming tracks its own read position with offsets stored in a checkpoint, so a restarted query resumes where it left off.

What should you look for when ingesting Kafka with Python?

Capability	Why it matters	How it works on Databricks
Kafka connection	Subscribe to the right topics	kafka format with kafka.bootstrap.servers plus subscribe, subscribePattern or assign. Authentication includes Unity Catalog service credentials and SASL/SSL
Delivery guarantees	Restarts must not lose or double-count data	Checkpoints store offsets and commits. Replayable sources and idempotent sinks give exactly-once semantics
Python transformations	Reuse team skills and libraries	PySpark DataFrame API, from_json, and transformWithState for custom state
Late events	Event time rarely matches arrival time	withWatermark, windowed aggregations and session_window
Rate control	Absorb bursts and backlogs	maxOffsetsPerTrigger limits the offsets processed per trigger interval
Monitoring	Know when the stream falls behind	Spark UI Streaming tab, StreamingQueryListener, Kafka lag metrics such as avgOffsetsBehindLatest, and Jobs notifications for streaming backlog (not for pipeline tasks)
Serving results	Dashboards and apps need fast reads	Unity Catalog tables, queried with Lakehouse//RT (Beta) when you need low-latency reads

How does Databricks handle Kafka ingestion and Python transformation?

This pipeline reads a topic, parses the payload and counts events per window:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, from_json, window

@dp.table
def kafka_raw():
    return (
        spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", "kafka_server:9092")
        .option("subscribe", "topic1")
        .load()
    )

@dp.table
@dp.expect_or_drop("has_user", "user_id IS NOT NULL")
def events():
    schema = "user_id STRING, event_type STRING, event_ts TIMESTAMP"
    return (
        spark.readStream.table("kafka_raw")
        .select(from_json(col("value").cast("string"), schema).alias("value"))
        .select("value.*")
    )

@dp.table
def events_per_window():
    return (
        spark.readStream.table("events")
        .withWatermark("event_ts", "10 minutes")
        .groupBy(window("event_ts", "5 minutes"), "event_type")
        .count()
    )

Databricks notes that aggregations are processed incrementally in pipelines only when you use watermarks. For an always-on stream, Databricks recommends running the pipeline with a continuous job, and continuous jobs retry a failed run with exponential backoff. On serverless compute, the Kafka docs recommend the availableNow trigger for incremental streaming and Lakeflow Pipelines continuous mode for low-latency continuous streaming.
The default micro-batch trigger fits analytical processing where latency is measured in seconds or minutes. When you need sub-second processing, Databricks documents real-time mode, a Structured Streaming trigger type that supports Python and supports Kafka as a source and a sink. Lakehouse Real-Time (Lakehouse//RT) is separate. It is a serverless SQL warehouse type in Beta that runs read-only queries over Unity Catalog tables, including streaming tables. A workspace admin enables the Beta, and the docs say its feature set will change before general availability.

What common Kafka patterns can you build in Python?

• Enrichment: A stream-static join combines each micro-batch with the latest valid version of a static Delta table. The join is stateless, but the output isn't deterministic if the static table changes.
• Sessions: Group events with session_window after withWatermark.
• Custom state: Use transformWithState, which Spark provides in Python from version 4.0.
• Operational decisions: Databricks cites flagging a card transaction when a fraud score exceeds a threshold as a real-time mode use case.

FAQs

What latency should I expect from Kafka to Delta?

It depends on the trigger. The default micro-batch trigger targets analytical processing measured in seconds or minutes. For sub-second end-to-end processing use real-time mode, and benchmark it against your own workload to choose a trigger interval.

Can one query read several topics?

Yes. subscribe takes a comma-separated list, subscribePattern takes a Java regex, and assign takes specific partitions. Each row has a topic column, so you can branch logic per topic.

How do I handle late events?

Define a watermark with withWatermark. Spark guarantees that data delayed by less than the threshold is aggregated, while data delayed by more may or may not be processed. A smaller threshold emits results sooner and drops more late records.

Will I get duplicate records?

Possibly. Structured Streaming does not automatically de-duplicate records from sources, and many message queues have at-least-once guarantees. Use dropDuplicatesWithinWatermark() with a watermark to remove them.