This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 5471de05232d [SPARK-57234][4.2][SS][DOC] Add Real-time Mode
documentation page to the Structured Streaming guide
5471de05232d is described below
commit 5471de05232d1082fee50a1d5d9424c81215600a
Author: Boyang Jerry Peng <[email protected]>
AuthorDate: Fri Jul 17 11:49:48 2026 -0700
[SPARK-57234][4.2][SS][DOC] Add Real-time Mode documentation page to the
Structured Streaming guide
### What changes were proposed in this pull request?
This is a backport of #56314 to `branch-4.2`.
This PR adds a new documentation page for **Real-time Mode** in Structured
Streaming, introduced in Spark 4.1.0 (SPARK-53736):
`docs/streaming/real-time-mode.md`. The page covers:
- **How Real-time Mode works**: long-running tasks (one per input
partition) that process records continuously, in contrast to the
per-micro-batch task scheduling of the default engine.
- **Batch duration is a checkpoint interval, not a latency target**.
- A **comparison** with the other execution modes.
- **Enabling Real-time Mode**: the `Trigger.RealTime(...)` API (Scala/Java)
and the `realTime` trigger keyword argument (Python), plus the requirements to
start (update output mode, checkpoint location, minimum batch duration).
- **Supported queries** (stateless only), **fault tolerance** (exactly-once
processing semantics; sinks such as Kafka provide at-least-once delivery),
**configuration**, **examples** (Python/Scala/Java), and **caveats**.
It also registers the new page in the Structured Streaming left navigation
(`docs/_data/menu-streaming.yaml`).
### Why are the changes needed?
Real-time Mode (stateless) was added in Spark 4.1.0 but has no user-facing
documentation in the Structured Streaming programming guide. This PR adds that
page. See SPARK-57234.
### Does this PR introduce _any_ user-facing change?
No. This is a documentation-only change.
### How was this patch tested?
Documentation-only change. The new page was validated for structure (front
matter, code tabs, Liquid `{% highlight %}` tags), internal links and in-page
anchors, navigation-menu anchors, and ASCII-only content. All trigger API
signatures, configuration keys and defaults, the supported-operator and sink
lists, and error-class references were cross-checked against the Spark 4.1.0
source (`Trigger.java`, `Triggers.scala`, `RealTimeModeAllowlist.scala`,
`SQLConf.scala`, `KafkaMicroBatchSt [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57325 from jerrypeng/backport-SPARK-57234-branch-4.2.
Authored-by: Boyang Jerry Peng <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
docs/_data/menu-streaming.yaml | 23 ++
docs/streaming/real-time-mode.md | 488 +++++++++++++++++++++++++++++++++++++++
2 files changed, 511 insertions(+)
diff --git a/docs/_data/menu-streaming.yaml b/docs/_data/menu-streaming.yaml
index b1dd02445112..7f01bb0c28cd 100644
--- a/docs/_data/menu-streaming.yaml
+++ b/docs/_data/menu-streaming.yaml
@@ -39,6 +39,29 @@
url:
streaming/apis-on-dataframes-and-datasets.html#recovering-from-failures-with-checkpointing
- text: Recovery Semantics after Changes in a Streaming Query
url:
streaming/apis-on-dataframes-and-datasets.html#recovery-semantics-after-changes-in-a-streaming-query
+- text: Real-time Mode
+ url: streaming/real-time-mode.html
+ subitems:
+ - text: How Real-time Mode Works
+ url: streaming/real-time-mode.html#how-real-time-mode-works
+ - text: Batch Duration Is a Checkpoint Interval
+ url:
streaming/real-time-mode.html#batch-duration-is-a-checkpoint-interval
+ - text: Comparison with Other Modes
+ url: streaming/real-time-mode.html#comparison-with-other-modes
+ - text: Enabling Real-time Mode
+ url: streaming/real-time-mode.html#enabling-real-time-mode
+ - text: Supported Queries
+ url: streaming/real-time-mode.html#supported-queries
+ - text: Fault Tolerance
+ url: streaming/real-time-mode.html#fault-tolerance
+ - text: Examples
+ url: streaming/real-time-mode.html#examples
+ - text: Configuration
+ url: streaming/real-time-mode.html#configuration
+ - text: Best Practices
+ url: streaming/real-time-mode.html#best-practices
+ - text: Caveats
+ url: streaming/real-time-mode.html#caveats
- text: Performance Tips
url: streaming/performance-tips.html
subitems:
diff --git a/docs/streaming/real-time-mode.md b/docs/streaming/real-time-mode.md
new file mode 100644
index 000000000000..15731dfb36a7
--- /dev/null
+++ b/docs/streaming/real-time-mode.md
@@ -0,0 +1,488 @@
+---
+layout: global
+displayTitle: Structured Streaming Programming Guide
+title: Structured Streaming Programming Guide
+license: |
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+---
+
+* Table of contents
+{:toc}
+
+# Real-time Mode
+
+**Real-time Mode** is a new streaming execution mode introduced in Spark 4.1.0
that
+targets ultra-low end-to-end latency with the exact same API and processing
guarantees / semantics as the current structured streaming engine.
+It is intended for operational workloads
+that must react to data the moment it arrives, such as fraud detection,
real-time alerting, and live
+personalization.
+
+In this release, Real-time Mode in Apache Spark supports **stateless queries**
only -- projections,
+filters and other map-like operations, unions, and stream-static joins.
Stateful operations such as
+streaming aggregations, deduplication, stream-stream joins, and
`transformWithState` are not yet
+supported, but support for them is planned starting in Spark 4.3. See
+[Supported Queries](#supported-queries) for the full list.
+
+The most important thing to know: **the duration you pass to the trigger
(default 5 minutes) is a
+checkpoint interval, not a latency target.** Records are processed and emitted
continuously rather
+than at batch boundaries, so the trigger duration does not set latency the way
a micro-batch interval does. See
+[Batch Duration Is a Checkpoint
Interval](#batch-duration-is-a-checkpoint-interval).
+
+You enable Real-time Mode by setting a Real-time trigger on the streaming
write; the rest of your
+query is unchanged. See [Enabling Real-time Mode](#enabling-real-time-mode).
+
+## How Real-time Mode Works
+
+By default, Structured Streaming runs a query as a series of small batch jobs
-- the *micro-batch*
+model. For each micro-batch, the driver plans the batch and launches a fresh
set of short-lived
+tasks. Those tasks read and process a bounded slice of the input, and the
driver commits progress
+before planning the next batch. The fixed per-batch planning and
task-scheduling overhead places a
+floor on end-to-end latency.
+
+Real-time Mode removes this per-batch overhead by launching **long-running
tasks** -- one per input
+partition. These tasks stay alive for the duration of a (long) batch and
process records
+continuously as they arrive. Because tasks are scheduled once per batch rather
than once per slice
+of data, records flow through the operator pipeline (source -> transformations
-> sink) without
+waiting for a batch boundary. End-to-end latency drops from the ~100 ms
micro-batch floor to roughly
+the time needed to process and ship one record (often a few milliseconds).
+
+Since records never wait for a batch boundary, the batch duration mainly
controls how often the
+query checkpoints progress -- as the next section explains.
+
+## Batch Duration Is a Checkpoint Interval
+
+In Real-time Mode, the batch duration is a **checkpoint interval, not a
latency interval.** With the
+default 5-minute duration, the query still emits records within milliseconds;
the 5 minutes only
+controls how often it commits progress and starts the next long-running batch.
This is the opposite
+of the micro-batch engine, where a longer batch interval directly increases
latency.
+
+Do not confuse the 5-minute default trigger duration with the 5-second minimum
allowed duration
+described under [Requirements](#requirements): the former is the checkpoint
cadence used when you do
+not specify a duration, while the latter is the smallest duration you are
allowed to set.
+
+Choosing the batch duration is a trade-off:
+
+- A *shorter* batch duration checkpoints more often, giving finer-grained
recovery (less work to
+ re-process after a failure). However, the query does not process data while
it commits progress
+ and starts the next batch, so checkpointing too frequently adds more of
these gaps, which can
+ raise tail (p99) latency, in addition to incurring more planning and commit
overhead.
+- A *longer* batch duration checkpoints less often, reducing that overhead and
those gaps, at the
+ cost of coarser-grained recovery (more data re-processed after a failure).
+
+The duration is set on the Real-time trigger, as shown under
+[Enabling Real-time Mode](#enabling-real-time-mode).
+
+## Comparison with Other Modes
+
+The table below summarizes how Real-time Mode relates to the default
micro-batch engine and to the
+experimental [Continuous
Processing](./performance-tips.html#continuous-processing) mode. See
+[How Real-time Mode Works](#how-real-time-mode-works) for the mechanism and
+[Supported Queries](#supported-queries) for the full list of supported
operations.
+
+| Mode | Latency | Processing Guarantees | Supported operations | When to use |
+|---|---|---|---|---|
+| Micro-batch (default) | ~100 ms | Exactly-once | All streaming operations,
including stateful | Stateful or higher-throughput workloads, or queries
Real-time Mode does not yet support |
+| Real-time Mode | millisecond-scale | Exactly-once | Stateless today
(map-like operations, unions, and stream-static joins); designed to support all
query shapes, including stateful | Low-latency workloads |
+| Continuous Processing (experimental) | ~1 ms | At-least-once | Map-like only
(projections and selections); no stateful operations | Legacy; use Real-time
Mode instead |
+
+The **Processing Guarantees** column refers to processing semantics, defined
under
+[Fault Tolerance](#fault-tolerance); end-to-end delivery additionally depends
on the sink and is
+independent of the execution mode.
+
+Real-time Mode and Continuous Processing both target millisecond-scale
latency, but they differ
+substantially:
+
+- **Continuous Processing** (introduced in Spark 2.3) is, and remains,
experimental. It supports
+ only map-like operations -- projections and selections -- with no stateful
operations such as
+ aggregations or joins, and it provides at-least-once guarantees. Because it
is stateless, the
+ exactly-once *processing* guarantee discussed under [Fault
Tolerance](#fault-tolerance) does not
+ apply to it. These constraints have limited its adoption.
+- **Real-time Mode** is designed to support all query shapes, including
stateful operations, while
+ reusing Spark's mature components such as state management, the Catalyst
optimizer, and the
+ existing SQL operators. It provides exactly-once processing semantics. It
currently supports
+ stateless queries, with stateful support planned starting in Spark 4.3.
+
+For new low-latency workloads, prefer Real-time Mode over Continuous
Processing.
+
+## Enabling Real-time Mode
+
+To run a supported query in Real-time Mode, set a **Real-time trigger** on the
streaming write.
+Everything else in the query stays the same. For example, the following query
reads from a Kafka
+topic, applies a stateless transformation, and writes the result to another
Kafka topic. Records
+flow through with low latency even though the trigger is 5 minutes.
+
+<div class="codetabs">
+
+<div data-lang="python" markdown="1">
+{% highlight python %}
+spark \
+ .readStream \
+ .format("kafka") \
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
+ .option("subscribe", "input-topic") \
+ .load() \
+ .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
+ .writeStream \
+ .format("kafka") \
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
+ .option("topic", "output-topic") \
+ .option("checkpointLocation", "/path/to/checkpoint") \
+ .outputMode("update") \
+ .trigger(realTime="5 minutes") \
+ .start()
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+import org.apache.spark.sql.streaming.Trigger
+
+spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "input-topic")
+ .load()
+ .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+ .writeStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("topic", "output-topic")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .outputMode("update")
+ .trigger(Trigger.RealTime("5 minutes")) // enable Real-time Mode
+ .start()
+{% endhighlight %}
+</div>
+
+<div data-lang="java" markdown="1">
+{% highlight java %}
+import org.apache.spark.sql.streaming.Trigger;
+
+spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "input-topic")
+ .load()
+ .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+ .writeStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("topic", "output-topic")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .outputMode("update")
+ .trigger(Trigger.RealTime("5 minutes")) // enable Real-time Mode
+ .start();
+{% endhighlight %}
+</div>
+
+</div>
+
+### Trigger API
+
+- *Scala and Java*: the trigger is `Trigger.RealTime(...)`, imported from
+ `org.apache.spark.sql.streaming.Trigger`. Several forms are available:
+ + `Trigger.RealTime()` -- uses the default batch duration of 5 minutes.
+ + `Trigger.RealTime("5 minutes")` -- a duration string.
+ + `Trigger.RealTime(300000)` -- the batch duration in milliseconds, as a
`long`.
+ + `Trigger.RealTime(5, TimeUnit.MINUTES)` -- a value together with a
+ `java.util.concurrent.TimeUnit`.
+ + `Trigger.RealTime(Duration("10 seconds"))` -- a Scala
`scala.concurrent.duration.Duration`.
+- *Python*: pass the batch duration as a string to the `realTime` keyword
argument of `trigger()`,
+ for example `.trigger(realTime="5 minutes")`. The duration is required in
Python.
+
+### Requirements
+
+A query must satisfy all of the following before it can start in Real-time
Mode; each is checked when
+the query starts:
+
+- The output mode must be `update`. Any other output mode fails to start with
+ `STREAMING_REAL_TIME_MODE.OUTPUT_MODE_NOT_SUPPORTED`.
+- A `checkpointLocation` is required, as with any other Structured Streaming
query.
+- The batch duration must be at least
`spark.sql.streaming.realTimeMode.minBatchDuration`
+ (5000 ms, i.e. 5 seconds, by default); a shorter interval fails to start with
+ `INVALID_STREAMING_REAL_TIME_MODE_TRIGGER_INTERVAL`. The duration string
must parse to a positive
+ interval, and month-based intervals (for example, `"1 month"`) are not
accepted. (This 5-second
+ minimum is distinct from the 5-minute default; see
+ [Batch Duration Is a Checkpoint
Interval](#batch-duration-is-a-checkpoint-interval).)
+
+## Supported Queries
+
+Real-time Mode supports stateless, map-like queries only.
+
+The following operations, sources, and sinks are supported as of Spark 4.1.0:
+
+- *Operations*: stateless, map-like operations are supported:
+ + Projections: `select`, `selectExpr`, `withColumn`, `drop`, and the typed
`map` / `flatMap` Dataset operations.
+ + Selections: `where` / `filter`.
+ + Expressions that compile to a projection -- including functions such as
`from_json` / `to_json`
+ and scalar user-defined functions (UDFs).
+ + Column generators such as `explode`.
+ + `union` of two or more *distinct* streaming sources. Referencing the same
source DataFrame more
+ than once is not supported and fails with
+ `STREAMING_REAL_TIME_MODE.IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED`;
create a separate DataFrame
+ for each source instead.
+ + Stream-static joins, where a streaming DataFrame is joined with a static
DataFrame. The static
+ side must be broadcast (use the `broadcast(...)` hint), because Real-time
Mode does not support
+ shuffles.
+ + `withWatermark` (event-time watermark declaration) is allowed, although it
has no effect because
+ stateful operators are not supported. This lets queries that already
declare a watermark run in
+ Real-time Mode without modification.
+
+- *Sources*: the source must support Real-time Mode. In Apache Spark, the
**Kafka** source supports
+ Real-time Mode. An unsupported source fails with
+ `STREAMING_REAL_TIME_MODE.INPUT_STREAM_NOT_SUPPORTED`. (The built-in `rate`
source is not supported
+ as a Real-time source.)
+
+- *Sinks*:
+ + **Kafka** sink.
+ + **Foreach** sink (via `ForeachWriter`), for writing to arbitrary external
systems one record at
+ a time. See [Using
Foreach](./apis-on-dataframes-and-datasets.html#using-foreach-and-foreachbatch).
+ Note that `foreachBatch` is *not* supported, because it processes each
batch as a whole rather
+ than one record at a time.
+ + **Console** and **memory** sinks, which are useful for development and
debugging.
+
+ Other sinks fail with `STREAMING_REAL_TIME_MODE.SINK_NOT_SUPPORTED`.
+
+The operators and sinks used by a Real-time query are checked against an
allowlist before the query
+starts; anything outside the allowlist fails with
+`STREAMING_REAL_TIME_MODE.OPERATOR_OR_SINK_NOT_IN_ALLOWLIST`.
+
+### Not supported
+
+Stateful operations of any kind are not supported in this release. This
includes streaming
+aggregations, `dropDuplicates` / `dropDuplicatesWithinWatermark`,
stream-stream joins, `repartition`
+and other operations that introduce a shuffle, and stateful operators such as
+`flatMapGroupsWithState` and `transformWithState`. Support for stateful
operations is planned
+starting in Spark 4.3. Asynchronous progress tracking is also not supported;
enabling it fails with
+`STREAMING_REAL_TIME_MODE.ASYNC_PROGRESS_TRACKING_NOT_SUPPORTED`.
+
+## Fault Tolerance
+
+Real-time Mode provides the same **exactly-once processing** guarantees as the
default micro-batch
+engine. Two distinct guarantees are worth separating:
+
+- **Exactly-once processing** means every input record's effect on the state
the engine manages (for
+ example, aggregation counts) is applied effectively once, even across
failures and restarts.
+- **Delivery semantics** describe whether a record may be written to the
external system more than
+ once. This is a property of the **sink**, not the execution mode.
+
+Real-time Mode is exactly-once with respect to processing. End-to-end delivery
depends on the sink: a
+sink that performs idempotent or transactional writes can deliver
**exactly-once**, while other sinks
+deliver **at-least-once** (duplicates are possible after a failure). The
built-in Kafka sink provides
+at-least-once delivery, with or without Real-time Mode. Real-time Mode does
not yet ship an
+exactly-once sink, though one can be implemented.
+
+Internally, offsets are committed at the *end* of each batch, after the
corresponding records have
+already been written to the sink. If a query fails partway through a batch, it
resumes from the last
+committed offsets on restart and may re-write records emitted before the
failure. Design sinks to
+tolerate duplicates -- for example, with idempotent writes -- where
exactly-once output matters.
+
+## Examples
+
+The following examples read from Kafka and assume a running Kafka cluster.
Each example shows the
+same query in Python, Scala, and Java.
+
+### Stream-static join
+
+Enrich a stream by joining it with a static reference dataset. The static side
is wrapped in
+`broadcast(...)` so the join is executed as a broadcast (map-side) join, which
avoids a shuffle.
+
+<div class="codetabs">
+
+<div data-lang="python" markdown="1">
+{% highlight python %}
+from pyspark.sql.functions import broadcast
+
+# Static reference data, read once as a batch DataFrame.
+reference = spark.read.format("parquet").load("/path/to/reference")
+
+spark \
+ .readStream \
+ .format("kafka") \
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
+ .option("subscribe", "input-topic") \
+ .load() \
+ .selectExpr("CAST(key AS STRING) AS joinKey", "CAST(value AS STRING) AS
value") \
+ .join(broadcast(reference), "joinKey") \
+ .writeStream \
+ .format("kafka") \
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
+ .option("topic", "output-topic") \
+ .option("checkpointLocation", "/path/to/checkpoint") \
+ .outputMode("update") \
+ .trigger(realTime="5 minutes") \
+ .start()
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+import org.apache.spark.sql.functions.broadcast
+import org.apache.spark.sql.streaming.Trigger
+
+// Static reference data, read once as a batch DataFrame.
+val reference = spark.read.format("parquet").load("/path/to/reference")
+
+spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "input-topic")
+ .load()
+ .selectExpr("CAST(key AS STRING) AS joinKey", "CAST(value AS STRING) AS
value")
+ .join(broadcast(reference), "joinKey")
+ .writeStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("topic", "output-topic")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .outputMode("update")
+ .trigger(Trigger.RealTime("5 minutes"))
+ .start()
+{% endhighlight %}
+</div>
+
+<div data-lang="java" markdown="1">
+{% highlight java %}
+import static org.apache.spark.sql.functions.broadcast;
+import org.apache.spark.sql.streaming.Trigger;
+
+// Static reference data, read once as a batch DataFrame.
+Dataset<Row> reference =
spark.read().format("parquet").load("/path/to/reference");
+
+spark
+ .readStream()
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "input-topic")
+ .load()
+ .selectExpr("CAST(key AS STRING) AS joinKey", "CAST(value AS STRING) AS
value")
+ .join(broadcast(reference), "joinKey")
+ .writeStream()
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("topic", "output-topic")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .outputMode("update")
+ .trigger(Trigger.RealTime("5 minutes"))
+ .start();
+{% endhighlight %}
+</div>
+
+</div>
+
+### Writing to the console for development
+
+The console sink prints output to the driver's standard output and is handy
while developing a
+query. Note that the console sink buffers each batch's rows and prints them
when the batch commits,
+so its output appears once per batch -- here, every 30 seconds -- rather than
continuously. This
+makes it useful for inspecting results, but it does not reflect Real-time
Mode's true per-record
+latency; to observe that, use a row-by-row sink such as Kafka. A shorter batch
duration simply makes
+the console refresh more often.
+
+<div class="codetabs">
+
+<div data-lang="python" markdown="1">
+{% highlight python %}
+spark \
+ .readStream \
+ .format("kafka") \
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
+ .option("subscribe", "input-topic") \
+ .load() \
+ .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
+ .where("value IS NOT NULL") \
+ .writeStream \
+ .format("console") \
+ .option("checkpointLocation", "/path/to/checkpoint") \
+ .outputMode("update") \
+ .trigger(realTime="30 seconds") \
+ .start()
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+import org.apache.spark.sql.streaming.Trigger
+
+spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "input-topic")
+ .load()
+ .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+ .where("value IS NOT NULL")
+ .writeStream
+ .format("console")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .outputMode("update")
+ .trigger(Trigger.RealTime("30 seconds"))
+ .start()
+{% endhighlight %}
+</div>
+
+<div data-lang="java" markdown="1">
+{% highlight java %}
+import org.apache.spark.sql.streaming.Trigger;
+
+spark
+ .readStream()
+ .format("kafka")
+ .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
+ .option("subscribe", "input-topic")
+ .load()
+ .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+ .where("value IS NOT NULL")
+ .writeStream()
+ .format("console")
+ .option("checkpointLocation", "/path/to/checkpoint")
+ .outputMode("update")
+ .trigger(Trigger.RealTime("30 seconds"))
+ .start();
+{% endhighlight %}
+</div>
+
+</div>
+
+## Configuration
+
+| Configuration | Default | Meaning |
+|---|---|---|
+| `spark.sql.streaming.realTimeMode.minBatchDuration` | `5000` (ms, 5 seconds)
| The minimum batch duration, in milliseconds, allowed for a Real-time trigger.
See the batch-duration requirement under [Requirements](#requirements). |
+| `spark.sql.streaming.realTimeMode.allowlistCheck` | `true` | Whether to
verify that all operators and sinks used by a Real-time query are in the
supported allowlist. Disabling this check (not recommended) lets unsupported
operators and sinks run at your own risk. |
+
+## Best Practices
+
+- Real-time Mode launches long-running tasks -- one per input partition --
that continuously read,
+ process, and write data. The number of tasks a query needs depends on how
many partitions it reads
+ from its sources in parallel. Before starting a Real-time query, ensure the
cluster has enough
+ cores to run all of these tasks simultaneously and continuously. For
example, reading from a Kafka
+ topic with 10 partitions requires at least 10 cores for the query to make
progress. Real-time Mode
+ uses a fixed 1:1 mapping between Kafka topic partitions and reader tasks;
the `minPartitions`
+ option is not supported in Real-time Mode.
+- Run a single Real-time query per cluster. Because Real-time Mode holds its
task slots for the
+ entire batch duration, any other queries sharing the cluster compete for the
same slots, which can
+ starve the Real-time query of resources and increase its latency.
+
+## Caveats
+
+- Real-time Mode provides exactly-once processing semantics, but sinks may
receive duplicate records
+ after a failure. See [Fault Tolerance](#fault-tolerance) for how to design
sinks for exactly-once
+ writes.
+- Adaptive Query Execution (AQE) is not supported for Real-time Mode queries.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
