Streaming¶

spark_rest_api_reader also works as a streaming source via spark.readStream. Rather than pulling everything once, an incremental read is driven by a cursor field in the API's own response (e.g. updated_at) — each micro-batch asks the API for only what's changed since the last value it saw. This page covers the cursor options, how a micro-batch is executed, the string-vs-numeric comparison pitfall, and the checkpoint/ restart-safety caveat on older pyspark versions.

Cursor options¶

Option	Default	Description
`stream_cursor_field`	—	Dot path to the high-water-mark field in each returned record (e.g. `updated_at`, or `meta.updated_at` for a nested field).
`stream_cursor_param`	—	Query param name sent upstream with the last-seen cursor value (e.g. `updated_since`).
`stream_initial_value`	`""`	Starting cursor value for the very first micro-batch.
`stream_cursor_is_numeric`	`false`	Set `true` if `stream_cursor_field` is a raw (non-zero-padded) number — see Cursor comparison below.

stream_cursor_field uses the same dot-path syntax as records_path and next_cursor_path (see Response formats), so a cursor nested inside the record — meta.updated_at, say — works the same as a top-level one.

from pyspark.sql.types import (
    IntegerType,
    StringType,
    StructField,
    StructType,
    TimestampType,
)

from spark_rest_api_reader import RestApiDataSource

spark.dataSource.register(RestApiDataSource)

items_schema = StructType(
    [
        StructField("id", IntegerType()),
        StructField("name", StringType()),
        StructField("status", StringType()),
        StructField("updated_at", TimestampType()),
    ]
)

stream_df = (
    spark.readStream.format("rest_api")
    .schema(items_schema)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "api_key")
    .option("api_key_value", dbutils.secrets.get("api-scope", "api-key"))
    .option("api_key_name", "X-Api-Key")
    .option("pagination_type", "page_number")
    .option("stream_cursor_field", "updated_at")
    .option("stream_cursor_param", "updated_since")
    .option("stream_initial_value", "1970-01-01T00:00:00Z")
    .load()
)

(
    stream_df.writeStream.format("delta")
    .option("checkpointLocation", "/Volumes/main/bronze/_checkpoints/items")
    .trigger(processingTime="5 minutes")
    .toTable("main.bronze.items_raw")
)

As with the batch reader, resolve any secret with dbutils.secrets.get(...) on the driver before passing it into .option(...) — see Secrets handling in Getting started.

How a micro-batch runs¶

Each micro-batch pulls everything matching stream_cursor_field > last value, sending the last-seen value upstream as the stream_cursor_param query parameter. Whatever pagination_type is configured still applies within that pull — a micro-batch can walk multiple pages via page_number, cursor, offset_limit, or link_header exactly as a batch read would.

All of this happens within a single partition per micro-batch — there is no intra-batch fan-out the way partitions or pages fans out a batch read. This keeps cursor tracking simple and correct: there's only ever one "last-seen value" to reason about. If you need parallelism, the batch reader with partitions on a schedule is the better fit than streaming — see Pagination & partitioning.

Cursor comparison¶

By default, "greater than" is evaluated as a string comparison on stream_cursor_field values. That's correct for:

ISO-8601 timestamps ("2024-01-02T00:00:00Z" > "2024-01-01T00:00:00Z")
Zero-padded numeric IDs ("0010" > "0009")

But it's wrong for raw, non-zero-padded numeric cursors once they cross a digit-count boundary — lexicographically, "9" > "10" is True, because "9" sorts after "1" character-by-character. If your cursor field is a plain auto-increment integer (a row ID, say) rather than a zero-padded string, set:

.option("stream_cursor_is_numeric", "true")

to compare cursor values numerically instead. This only changes how values are compared; it does not change how the cursor value is sent upstream via stream_cursor_param — that's still whatever string representation the field naturally has.

from pyspark.sql.types import IntegerType, LongType, StringType, StructField, StructType

stream_df = (
    spark.readStream.format("rest_api")
    .schema(
        StructType(
            [
                StructField("id", IntegerType()),
                StructField("name", StringType()),
                StructField("row_id", LongType()),
            ]
        )
    )
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "api_key")
    .option("api_key_value", dbutils.secrets.get("api-scope", "api-key"))
    .option("stream_cursor_field", "row_id")
    .option("stream_cursor_param", "since_id")
    .option("stream_initial_value", "0")
    .option("stream_cursor_is_numeric", "true")
    .load()
)

Restart/checkpoint safety requires pyspark >= 4.2¶

This is the most important caveat on this page, so it's worth spelling out in full.

Spark's Python Data Source streaming API offers two different signatures for the method a source implements to report the latest available offset:

A legacy, zero-argument latestOffset() — the source just reports "here's the latest offset," with no information about where the query last left off.
An admission-control latestOffset(start, limit) — added in pyspark 4.2.0 — which receives start, the offset restored from the checkpoint, as an argument.

spark_rest_api_reader implements the admission-control signature. That signature is the only one that actually receives the restored checkpoint offset after a query restart — whether that restart is a driver restart, a job retry, or a redeploy of the same streaming job pointed at the same checkpointLocation. With it, the reader picks up the cursor value Spark restored from the checkpoint and resumes exactly where the query left off, without reprocessing records already ingested.

On an older pyspark/DBR where only the legacy zero-arg latestOffset() convention exists, the reader still works — it doesn't fail or crash — but it can only track the cursor in an in-memory attribute of the running process, because that's all the legacy signature gives it access to. That in-memory value resets every time a fresh process starts. Concretely, that means:

The stream runs correctly across ordinary micro-batches within one continuously-running query.
But a restart — the process restarting for any reason — starts the cursor back over at stream_initial_value, and the query reprocesses everything already ingested from the beginning, rather than failing loudly.

That silent full-reprocessing is the risk: it degrades correctness (and cost, and downstream duplicate-handling burden) without an obvious error to point at, since nothing raises — the stream just quietly starts over.

If your streaming jobs need to survive restarts without full reprocessing, confirm the cluster's pyspark version is >= 4.2 before relying on this in production. On Databricks, that means a Databricks Runtime version that ships pyspark 4.2.0 or later. If you're on an older runtime and can't upgrade, treat this reader's streaming mode as non-restart-safe: downstream tables should be built to tolerate reprocessed duplicates (e.g. MERGE/upsert on a natural key) until the cluster is upgraded.

Next steps¶

Pagination & partitioning — the pagination_type each micro-batch uses to walk its own pages.
Response formats — dot-path syntax shared by stream_cursor_field, records_path, and next_cursor_path.
Schema — declaring a TimestampType/IntegerType column for the cursor field so it round-trips correctly.
Examples — examples/streaming_example.py walks through the default (string-comparison) case, the numeric-cursor alternative, and a nested-cursor-field alternative.