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.