topic Re: SDP continuous mode in Data Engineering

SDP continuous mode

lachu — Mon, 22 Jun 2026 20:21:42 GMT

Hi,

I was doing a POC and hence used open source spark and kafka in docket container and got it working. The sample code is ingesting data from kafka but it is running only in batch mode. Not able to continuously ingest the kafka stream

Question: Can we create streaming continuous pipeline using open source spark?

Thanks

Re: SDP continuous mode

bala_sai — Tue, 23 Jun 2026 02:55:37 GMT

Yes, we can build a continuous streaming pipeline using open source Spark. The main thing is to use Spark Structured Streaming, not a normal batch read. For Kafka streaming, we need to use spark.readStream, then write using writeStream, and keep the query running with awaitTermination().

Sample code (python):

# Read continuously from Kafka
kafka_df = (
spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "kafka:9092")
.option("subscribe", "your_topic_name")
.option("startingOffsets", "latest")
.load()
)

# Convert Kafka key/value from binary to string
parsed_df = (
kafka_df
.select(
col("key").cast("string").alias("key"),
col("value").cast("string").alias("value"),
col("timestamp")
)
)

# Write stream output
query = (
parsed_df.writeStream
.format("delta") # can be changed to parquet/delta etc.
.outputMode("append")
.option("checkpointLocation", "/tmp/checkpoints/kafka_stream_poc")
.start()
)

# Keeps the streaming job alive
query.awaitTermination()

Re: SDP continuous mode

lachu — Tue, 23 Jun 2026 04:29:13 GMT

Hmm. This looks more like imperative programming than sdp

Would you be able to give me a sample with @DP.table?

Re: SDP continuous mode

Yogasathyandrun — Tue, 23 Jun 2026 11:22:35 GMT

Yes, you can build a continuously running streaming pipeline on open-source Spark

Structured Streaming (imperative)

For Kafka, the standard approach is readStream → writeStream, kept alive with awaitTermination():

kafka_df = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("subscribe", "your_topic_name")
         .load()
)

query = (
    kafka_df.writeStream
            .format("delta")                              # or "console" for a quick POC check
            .outputMode("append")
            .option("checkpointLocation", "/tmp/checkpoints/kafka_stream")
            .option("path", "/tmp/output/kafka_stream")   # required for file sinks
            .start()
)
query.awaitTermination()

A POC usually behaves like a batch job for one of two reasons: it uses spark.read.format("kafka") (batch) instead of spark.readStream.format("kafka"), or it has a one-time trigger like .trigger(availableNow=True) (the newer form of .trigger(once=True)), which drains the topic once and stops. With readStream + writeStream + awaitTermination() and no one-time trigger, Spark keeps consuming new messages as they arrive.

Spark Declarative Pipelines (@dp.table)

The declarative equivalent — you define the datasets and let the framework handle execution, checkpointing, and orchestration:

from pyspark import pipelines as dp
from pyspark.sql.functions import col

# Bronze - raw Kafka stream
@dp.table(name="kafka_events")
def kafka_events():
    return (
        spark.readStream
             .format("kafka")
             .option("kafka.bootstrap.servers", "kafka:9092")
             .option("subscribe", "your_topic_name")
             .load()
             .select(
                 col("key").cast("string").alias("key"),
                 col("value").cast("string").alias("value"),
                 col("timestamp")
             )
    )

# Silver - simple transformation
@dp.table(name="events_clean")
def events_clean():
    return (
        spark.readStream.table("kafka_events")
             .filter(col("value").isNotNull())
    )

Note there's no writeStream.start() / awaitTermination() inside the definitions — the framework manages that.

Running an SDP pipeline with spark-pipelines run performs a triggered/incremental update in open-source Spark 4.1. The always-on continuous mode isn't in the open-source framework yet — it's a Databricks Lakeflow extension. So dp.table gives you the declarative model, but for a process that runs forever, the Structured Streaming version above is still the route.

If you can share the Kafka ingestion code you're currently using, and whether you're targeting Structured Streaming or the pyspark.pipelines SDP framework, it'd be easier to pinpoint why the current run behaves like a batch job.

Re: SDP continuous mode

lachu — Tue, 23 Jun 2026 15:07:25 GMT

Sample code that i used

from pyspark import pipelines as dp from pyspark.sql import DataFrame, SparkSession, functions as f from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType spark = SparkSession.active() @dp.temporary_view() def event_stream() -> DataFrame: """Stream order events from Kafka.""" schema = StructType([ StructField("event_id", IntegerType()), StructField("timestamp", StringType()), StructField("user_id", StringType()), StructField("event_type", StringType()), StructField("product", StringType()), StructField("price", DecimalType(8,2)), StructField("quantity", IntegerType()), ]) return ( spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "kafka:29092") .option("subscribe", "events") .option("startingOffsets", "earliest") .load() .select(f.from_json(f.col("value").cast("string"), schema).alias("data")) .select("data.*") ) # The @DP.table decorator combined with spark.readStream marks this as a streaming table (format='delta') def bronze_events() -> DataFrame: """Persist the stream into a Delta Bronze layer continuously.""" # Correct open-source SDP syntax to read an upstream view as a stream return spark.readStream.table('event_stream')

Sample config

name: StreamTest_001 storage: "file:////opt/spark/data/checkpoints/StreamTest_001" catalog: spark_catalog database: default libraries: - glob: include: "./stream.py" configuration: spark.remote: "sc://localhost:15002" spark.sql.shuffle.partitions: "2" pipelines.trigger.type: "ProcessingTime" pipelines.trigger.interval: "10 seconds"