From 7ae1f8bb1dc90309ef6ea05ee0591bf52248b0f6 Mon Sep 17 00:00:00 2001 From: "databricks-ci-ghec-1[bot]" <184311507+databricks-ci-ghec-1[bot]@users.noreply.github.com> Date: Thu, 30 Jul 2026 03:28:45 +0000 Subject: [PATCH] Update agent skills to ef1897c8f4570be9ef2b40e05944c12c8fd4872c --- .gen.json | 2 +- skills/databricks-pipelines/SKILL.md | 2 + .../references/pipeline-configuration.md | 4 +- .../references/real-time-mode.md | 153 ++++++++++++++++++ .../references/sink-python.md | 2 + .../references/streaming-patterns.md | 2 +- .../references/real-time-mode.md | 2 +- 7 files changed, 162 insertions(+), 5 deletions(-) create mode 100644 skills/databricks-pipelines/references/real-time-mode.md diff --git a/.gen.json b/.gen.json index 65a8a375..540d74d8 100644 --- a/.gen.json +++ b/.gen.json @@ -1,3 +1,3 @@ { - "source_commit": "007006859ac9df636ebfa08d5d25a1d5b59f15e6" + "source_commit": "ef1897c8f4570be9ef2b40e05944c12c8fd4872c" } diff --git a/skills/databricks-pipelines/SKILL.md b/skills/databricks-pipelines/SKILL.md index 35d7c153..6ea7340c 100644 --- a/skills/databricks-pipelines/SKILL.md +++ b/skills/databricks-pipelines/SKILL.md @@ -114,6 +114,7 @@ Some features sit on top of others — read both: | Backfill Flow | One-time historical load + ongoing live stream into same table. | `@dp.append_flow(once=True)` | `CREATE FLOW ... INSERT INTO ... ONCE` | [streaming-table-python](references/streaming-table-python.md) | [streaming-table-sql](references/streaming-table-sql.md) | | Sink (Delta/Kafka/EH/custom) | Write streaming output to external Delta / Kafka / Event Hubs. | `dp.create_sink()` | N/A — Python only | [sink-python](references/sink-python.md) | — | | ForEachBatch Sink | Custom per-batch Python logic (merge/upsert, multi-destination). Public Preview. | `@dp.foreach_batch_sink()` | N/A — Python only | [foreach-batch-sink-python](references/foreach-batch-sink-python.md) | — | +| RTM update flow | Real-Time Mode: route a flow to a sink with sub-second latency. Public Preview. | `@dp.update_flow(target=...)` | N/A — Python only | [real-time-mode](references/real-time-mode.md) | — | ### CDC APIs @@ -258,6 +259,7 @@ Cross-cutting patterns: - [streaming-patterns.md](references/streaming-patterns.md) — Dedup, windowed aggregations, late data, rescue-data quarantine, anomaly detection, lag monitoring. - [scd-2-querying.md](references/scd-2-querying.md) — Current-state, point-in-time, joining facts with historical dims. - [kafka.md](references/kafka.md) — Kafka / Event Hubs ingestion. +- [real-time-mode.md](references/real-time-mode.md) — Real-Time Mode (RTM): sub-second continuous pipelines (`@dp.update_flow`, Kafka sinks, serverless or classic). Auto Loader format-specific options: [JSON](references/options-json.md) · [CSV](references/options-csv.md) · [XML](references/options-xml.md) · [Parquet](references/options-parquet.md) · [Avro](references/options-avro.md) · [Text](references/options-text.md) · [ORC](references/options-orc.md). diff --git a/skills/databricks-pipelines/references/pipeline-configuration.md b/skills/databricks-pipelines/references/pipeline-configuration.md index 0be40d5b..2729b26d 100644 --- a/skills/databricks-pipelines/references/pipeline-configuration.md +++ b/skills/databricks-pipelines/references/pipeline-configuration.md @@ -29,7 +29,7 @@ databricks pipelines create --json '{ Per-field rationale: -- **`continuous: false`** — triggered runs. `true` auto-restarts failed updates forever (`cause: RETRY_ON_FAILURE`), burning cost and trapping polling loops. Only `true` when the user explicitly asks for always-on streaming. +- **`continuous: false`** — triggered runs. `true` auto-restarts failed updates forever (`cause: RETRY_ON_FAILURE`), burning cost and trapping polling loops. Only `true` when the user explicitly asks for always-on streaming. (Real-Time Mode is the one sanctioned always-on case — it *requires* `continuous: true`; see [real-time-mode.md](real-time-mode.md).) - **`development: true`** — faster startup, relaxed validation, no retry-on-failure. Required for any edit/re-run loop. - **`pipelines.numUpdateRetryAttempts: "0"` + `maxFlowRetryAttempts: "0"`** — belt-and-suspenders against retries. Even with `development`, some configs still retry. Drop for prod. - **`channel: "PREVIEW"`** — latest features. `"CURRENT"` (default) for production stability. @@ -130,7 +130,7 @@ All values must be strings. |-----|-------------| | `spark.sql.shuffle.partitions` | Number of shuffle partitions. `"auto"` recommended. | | `pipelines.numRetries` | Retries on transient failures. | -| `pipelines.trigger.interval` | Trigger interval for continuous pipelines (e.g. `"1 hour"`). | +| `pipelines.trigger.interval` | Trigger interval for continuous pipelines (e.g. `"1 hour"`). In a Real-Time Mode flow this same key instead sets the long-running batch's checkpoint cadence (e.g. `"5 minutes"`), not a trigger frequency — see [real-time-mode.md](real-time-mode.md). | | `spark.databricks.delta.preview.enabled` | Enable Delta preview features (`"true"`). | Any key here is also accessible from pipeline code via `spark.conf.get("key")` — use this to parameterize transformations. diff --git a/skills/databricks-pipelines/references/real-time-mode.md b/skills/databricks-pipelines/references/real-time-mode.md new file mode 100644 index 00000000..acfadfec --- /dev/null +++ b/skills/databricks-pipelines/references/real-time-mode.md @@ -0,0 +1,153 @@ +# Real-Time Mode (RTM) on Lakeflow Declarative Pipelines + +RTM runs an SDP flow on a continuous engine instead of micro-batches, targeting end-to-end latency "as low as five milliseconds" ([Use real-time mode in SDP](https://docs.databricks.com/aws/en/ldp/real-time)). You keep the declarative `sinks + flows` authoring surface — no `writeStream`, no `awaitTermination`, no checkpoint paths — and the framework runs it continuously. **Public Preview.** + +This file is SDP-specific. For standalone Structured Streaming RTM (`writeStream…trigger(realTime=…)` on classic compute) — including the shared error classes, cluster prohibitions, and observability internals — see [../databricks-spark-structured-streaming/references/real-time-mode.md](../../databricks-spark-structured-streaming/references/real-time-mode.md). + +## When to reach for RTM + +RTM pipelines run **continuously** (`continuous: true`) — the compute never scales to zero. That makes RTM materially more expensive than a triggered pipeline. Reach for it only for **operational use cases** with a real sub-second/low-hundreds-of-ms SLA (fraud scoring, live alerting, personalization). For demos, prototypes, or other use cases that tolerate seconds-to-minutes latency, use a normal triggered pipeline instead. Validate with the user before recommending RTM, and confirm they accept always-on compute. (This is the sanctioned exception to the general "avoid `continuous: true`" guidance in [pipeline-configuration.md](pipeline-configuration.md).) + +## SDP-on-RTM vs standalone RTM + +| | SDP-on-RTM (this file) | Standalone RTM (structured-streaming skill) | +|---|---|---| +| Authoring | `dp.create_sink` + `@dp.update_flow` | `writeStream…trigger(realTime=…)` | +| Lifecycle | Managed by the pipeline (orchestration, checkpoints, retries, state) | You manage the query | +| Compute | **Serverless or classic** | Classic only | +| Sinks | Kafka (Delta is not RTM-usable — see below) | Kafka + custom `foreach` + native Lakebase (Public Preview) | + +## Enabling RTM + +Three things together turn RTM on: + +1. **Pipeline settings** — `continuous: true`, `serverless: true` (or classic), on the `PREVIEW` channel. Requires **Databricks Runtime 18.1.3** on the SDP preview channel ([docs](https://docs.databricks.com/aws/en/ldp/real-time)). +2. **Pipeline-level Spark conf** — `spark.databricks.streaming.realTimeMode.enabled: true`. +3. **Per-flow trigger** — `pipelines.trigger: "RealTime"` in the `@dp.update_flow` `spark_conf` (see the pattern below). + +## The `dp.create_sink` + `@dp.update_flow` pattern + +RTM delivers to an operational system rather than a table, so you write to an external **sink** and route a **flow** to it. + +> **`target=` is a sink NAME, not a Kafka topic.** Per the [update_flow reference](https://docs.databricks.com/aws/en/ldp/developer/ldp-python-ref-update-flow), `target` is *"Required. The name of the sink this flow writes to."* Every `@dp.update_flow(target="X")` must be preceded by a `dp.create_sink("X", …)` that declares `X`. The Kafka connection lives in `create_sink`, not the decorator. + +```python +from pyspark import pipelines as dp + +# 1. Declare the sink. Its output DataFrame must have a `value` column; +# `key`, `partition`, `headers`, `topic` are optional. +dp.create_sink( + name="kafka_out_sink", + format="kafka", + options={ + "kafka.bootstrap.servers": "kafka-broker:9092", + "topic": "rtm_output", + }, +) + +# 2. Route a real-time flow to that sink NAME. +@dp.update_flow( + name="kafka_rtm_flow", + target="kafka_out_sink", # the sink declared above — NOT a topic + spark_conf={ + "pipelines.trigger": "RealTime", # turns RTM on for this flow + "pipelines.trigger.interval": "5 minutes", + }, +) +def kafka_rtm_flow(): + return ( + spark.readStream.format("kafka") + .option("kafka.bootstrap.servers", "kafka-broker:9092") + .option("subscribe", "raw_events") + .option("startingOffsets", "latest") + .load() + .selectExpr("CAST(key AS STRING) AS key", + "CAST(value AS STRING) AS value", + "timestamp") + ) +``` + +**Enrich with a broadcast stream-static join.** Broadcast is the only supported stream-static join shape in RTM; wrap the static side in `broadcast()` and keep it small enough to fit in memory: + +```python +from pyspark.sql.functions import broadcast + +dp.create_sink( + name="enriched_events_sink", + format="kafka", + options={"kafka.bootstrap.servers": "kafka-broker:9092", "topic": "enriched_events"}, +) + +@dp.update_flow( + name="enriched_events_flow", + target="enriched_events_sink", + spark_conf={"pipelines.trigger": "RealTime", "pipelines.trigger.interval": "5 minutes"}, +) +def enriched_events_flow(): + events = ( + spark.readStream.format("kafka") + .option("kafka.bootstrap.servers", "kafka-broker:9092") + .option("subscribe", "raw_events") + .load() + .selectExpr("CAST(key AS STRING) AS user_id", + "CAST(value AS STRING) AS value", + "timestamp") + ) + users = spark.read.table("main.dim.users") # small static UC table, broadcast into the stream + return ( + events.join(broadcast(users), "user_id", "left") + .selectExpr("user_id AS key", + "to_json(struct(*)) AS value", + "timestamp") + ) +``` + +**`create_sink` format is kafka-only for RTM.** The signature accepts *"either `kafka` or `delta`"* ([create_sink reference](https://docs.databricks.com/aws/en/ldp/developer/ldp-python-ref-sink)), but **Delta is not a supported RTM sink** (see Sources, sinks, and operators below), so within an RTM flow only `format="kafka"` works. Use `delta` sinks in non-RTM flows only. + +**Benchmarking / testing:** to run an RTM flow without a real destination, use a `noop` sink — `dp.create_sink(name="bench_sink", format="noop", options={})` discards output (it passes through to the Structured Streaming writer). It's not a documented SDP `create_sink` format and has no support guarantee — fine for local benchmarking, not production. + +## `pipelines.trigger.interval` — checkpoint cadence, not batch size + +In RTM the batch is long-running and records are processed as they arrive; `pipelines.trigger.interval` (default `"5 minutes"`) governs **how often state and source offsets are checkpointed**, not how often results appear. This is a different meaning from the same key on a normal continuous pipeline (where it's a coarse trigger like `"1 hour"` — see [pipeline-configuration.md](pipeline-configuration.md)). Keep it at minutes; shorter intervals add checkpoint overhead, longer ones increase replay-on-restart. + +## Compute + +- **Serverless** — managed for you; the default and simplest choice. +- **Classic** — set compute in the pipeline's `clusters` config. Keep **Photon off** (RTM does not use Photon, so enabling it only adds the Photon DBU uplift for no benefit) and **autoscaling off** (RTM runs continuously — size the cluster to fixed capacity). A pipeline will still *start* with either enabled, but neither helps an RTM flow, so leave them disabled. + - **Size for the slot math.** RTM schedules all stages concurrently, so free slots must cover the **sum of partitions across *every* stage**, not just the source: source read tasks **+** each stateful stage's `spark.sql.shuffle.partitions` **+** any explicit `repartition(n)`. E.g. an 8-task Kafka source feeding one `groupBy` at `shuffle.partitions=20` needs 8 + 20 = 28 slots. Undersize → the flow fails at start with `CONCURRENT_SCHEDULER_INSUFFICIENT_SLOT` (note: this class is **not** in the `STREAMING_REAL_TIME_MODE.*` namespace). Your two levers to shrink the total: cap the source with `maxPartitions` (reads topic partitions across fewer tasks; unset = topic partition count) and set `shuffle.partitions` low. RTM also allows **at most one streaming shuffle stage** per flow (`SHUFFLE_MORE_THAN_ONCE` otherwise) — combine aggregations or use a broadcast join to stay within one. The SS ref's [slot-math table](../../databricks-spark-structured-streaming/references/real-time-mode.md) works through more shapes. On **serverless** you don't size a cluster yourself — it scales to fit — but you should **still set these partition counts**. + +## Sources, sinks, and operators + +Sources/sinks and most operator restrictions are the **same as standalone RTM** — see its [reference and error-class matrix](../../databricks-spark-structured-streaming/references/real-time-mode.md) for the full set. Sources/sinks: Kafka / MSK / Event Hubs (Kafka connector) as source and sink, Kinesis (EFO) source-only; Delta and file-based sources (Auto Loader, direct file reads) are not supported. The same operators are unsupported as in standalone RTM — session windows, `dropDuplicatesWithinWatermark`, `flatMapGroupsWithState`, `mapPartitions`, and `transformWithStateInPandas`. + +**Where SDP-on-RTM is *more* restricted than standalone RTM:** + +- **Stream-to-stream joins** — standalone RTM added an inner stream-stream join on DBR 18+, but SDP-on-RTM does not support them. Use a broadcast stream-static join instead. +- **Custom `forEach` sinks** — standalone RTM supports a custom `ForeachWriter`; SDP does not (it exposes only `dp.create_sink` and `@dp.foreach_batch_sink`, with no way to supply a raw `ForeachWriter`). + +**RTM flows must be streaming.** A batch flow (materialized view, `spark.read`) can't be an RTM flow — it fails analysis with "real-time mode is only supported for streaming queries." Use a streaming read (`spark.readStream`). + +**Not applicable in RTM:** Auto CDC (`dp.create_auto_cdc_flow` / `apply_changes`) is rejected in an RTM flow — RTM flows are plain streaming reads written to a sink. + +`transformWithState` itself is supported, with RTM-specific behavior the [SDP docs](https://docs.databricks.com/aws/en/ldp/real-time) spell out: `handleInputRows` is invoked **once per row** (not once per key per batch) and **event-time timers are unsupported** (processing-time only). Same behavior as standalone RTM — see its [reference](../../databricks-spark-structured-streaming/references/real-time-mode.md) for deeper detail. + +## One real-time flow per pipeline + +Run **one real-time flow per pipeline, and keep non-RTM (micro-batch) flows out of it** — slots aren't reserved per flow, so a co-located flow (especially a bursty micro-batch one) contends for the slots the RTM flow needs. + +## Lakebase as a serving layer + +An RTM flow in SDP sinks to Kafka — there is no Lakebase sink in the SDP RTM path. To serve an app from Lakebase, wire it up outside the RTM flow; see the [databricks-lakebase](../../databricks-lakebase/SKILL.md) skill (e.g. synced tables). + +## Observability and tuning + +RTM emits per-batch latency percentiles (`processingLatencyMs`, `sourceQueuingLatencyMs`, `e2eLatencyMs`) in the streaming query progress; watch p99, not the average. See the [standalone RTM reference](../../databricks-spark-structured-streaming/references/real-time-mode.md) for what each measures. + +**Set `spark.sql.shuffle.partitions` on stateful flows.** The default is `200`, which a stateful stage (aggregation, `dropDuplicates`, stream-static join) turns into 200 concurrent slots — see the slot math above. Set it low, matched to the stage's real parallelism (the docs' aggregation example uses `"8"`), in the flow's `@dp.update_flow(spark_conf={...})`. + +## Related references + +- [sink-python.md](sink-python.md) — general SDP sinks (`dp.create_sink`, Delta/Kafka, `@dp.append_flow`). +- [kafka.md](kafka.md) — Kafka / Event Hubs source options. +- [streaming-patterns.md](streaming-patterns.md) — dedup, windowing, late data for non-RTM streaming. +- [../databricks-spark-structured-streaming/references/real-time-mode.md](../../databricks-spark-structured-streaming/references/real-time-mode.md) — standalone RTM: cluster setup, slot math, full error-class catalog, observability internals. diff --git a/skills/databricks-pipelines/references/sink-python.md b/skills/databricks-pipelines/references/sink-python.md index 4af8f860..f2691830 100644 --- a/skills/databricks-pipelines/references/sink-python.md +++ b/skills/databricks-pipelines/references/sink-python.md @@ -4,6 +4,8 @@ Sinks write pipeline output to non-pipeline-managed targets: Kafka / Event Hubs For per-batch custom Python logic (merge/upsert, multi-destination), see [foreach-batch-sink-python.md](foreach-batch-sink-python.md). +**Real-Time Mode:** in an RTM pipeline a sink is targeted by `@dp.update_flow` (not `@dp.append_flow`), and only `format="kafka"` is usable — Delta is not an RTM sink. See [real-time-mode.md](real-time-mode.md). + ## `dp.create_sink(...)` Call at top level before any `@dp.append_flow` references it. diff --git a/skills/databricks-pipelines/references/streaming-patterns.md b/skills/databricks-pipelines/references/streaming-patterns.md index d54bd34c..d158e470 100644 --- a/skills/databricks-pipelines/references/streaming-patterns.md +++ b/skills/databricks-pipelines/references/streaming-patterns.md @@ -91,7 +91,7 @@ Python: `F.session_window("event_timestamp", "30 minutes")`. | 15–60 minutes | Operational dashboards | | 1–24 hours | Analytical reports | -Larger windows = less state pressure but stale results. Pick the smallest window that meets the freshness SLO. +Larger windows = less state pressure but stale results. Pick the smallest window that meets the freshness SLO. For genuine sub-second operational latency, use [Real-Time Mode](real-time-mode.md). --- diff --git a/skills/databricks-spark-structured-streaming/references/real-time-mode.md b/skills/databricks-spark-structured-streaming/references/real-time-mode.md index 7ea06866..b3f4b4d1 100644 --- a/skills/databricks-spark-structured-streaming/references/real-time-mode.md +++ b/skills/databricks-spark-structured-streaming/references/real-time-mode.md @@ -18,7 +18,7 @@ RTM has hard cluster requirements. Get any of these wrong and the stream either | Setting | Required value | Notes | |---|---|---| | DBR | **16.4 LTS minimum, 18.1+ recommended** | Per the [GA blog](https://www.databricks.com/blog/announcing-general-availability-real-time-mode-apache-spark-structured-streaming-databricks): "we recommend DBR 18.1 for the latest features and optimizations." DBR 18+ is also required for stream-stream inner join (see [Supported operations](#supported-operations)) and DBR 18.2+ resolves a known latency floor with Python `transformWithState` at <5 rec/sec. | -| Compute type | **Classic compute** (Dedicated or Standard access mode) | Standard supports Python only. Serverless is NOT supported for standalone RTM — only inside SDP-on-RTM (see [sdp-real-time-mode.md](sdp-real-time-mode.md)). | +| Compute type | **Classic compute** (Dedicated or Standard access mode) | Standard supports Python only. Serverless is NOT supported for standalone RTM — only inside SDP-on-RTM (see [SDP real-time mode](../../databricks-pipelines/references/real-time-mode.md)). | | Autoscaling | **Off** | Streaming clusters must be fixed-size. | | Photon | **Off** | Incompatible with RTM. | | Spot instances | **Off** | Interruptions break the stream. |