Examples

The repository ships three example scripts under examples/. Each is written as a Databricks notebook/job snippet — they assume spark and dbutils are already in scope — and each is meant to be read top to bottom, then copied from and adapted rather than run as-is. This page walks through what each one demonstrates.

batch_example.py — batch pull into Delta

The main body of this script is the simplest possible production shape: a single explicit schema, bearer auth, page-number pagination, and an append into a Bronze Delta table.

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

from spark_rest_api_reader import RestApiDataSource

spark.dataSource.register(RestApiDataSource)

api_token = dbutils.secrets.get("api-scope", "api-token")

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

df = (
    spark.read.format("rest_api")
    .schema(ITEMS_SCHEMA)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "bearer")
    .option("token", api_token)
    .option("pagination_type", "page_number")
    .option("page_param", "page")
    .option("page_size_param", "page_size")
    .option("page_size", "200")
    .option("timeout", "30")
    .option("max_retries", "5")
    .load()
)

df.write.format("delta").mode("append").saveAsTable("main.bronze.items_raw")

Two things worth noticing before the alternatives below:

• api_token is resolved from dbutils.secrets.get(...) on the driver, then passed into .option(...) as a plain string. This is the only place a secret lookup can happen — options are shipped to executors as part of the reader, and dbutils isn't available on executors. See Getting started and Request & auth for more on this.
• The schema is a StructType, built with typed StructFields — not a DDL string and not a JSON-blob column. Nested response data always follows the same pattern; see Schema.

The rest of the file is a series of commented-out alternative blocks, each a self-contained variant of the same read. They cover:

Auto page-range fan-out instead of in-partition pagination — replace pagination_type with pages, and Spark generates one partition per page number up front for maximum parallelism, instead of one partition paginating sequentially:

df = (
    spark.read.format("rest_api")
    .schema(ITEMS_SCHEMA)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "bearer")
    .option("token", api_token)
    .option("pages", "20")
    .load()
)

Explicit partitioning by param overrides — one partition per region, each fully paginated on its own via cursor pagination, combined with oauth2_client_credentials auth:

.option("pagination_type", "cursor")
.option("next_cursor_path", "meta.next_cursor")
.option("partitions", '[{"region": "us"}, {"region": "eu"}, {"region": "apac"}]')

This is the example that shows partitioning and pagination composing: each of the three region partitions walks all of its own pages independently. See Pagination & partitioning for the general rule.

Offset/limit pagination (skip/take style APIs), paired with api_key auth:

.option("pagination_type", "offset_limit")
.option("offset_param", "skip")
.option("limit_param", "take")
.option("limit", "500")

RFC 5988 Link header pagination (GitHub/Stripe-style) — no extra options beyond the type itself; the reader follows rel="next" until the header is absent:

.option("pagination_type", "link_header")

Custom envelope shape with records_path and flatten — for a response like {"envelope": {"payload": [{"id": 1, "address": {"city": "NYC"}}, ...]}}, this shows records_path pointing at the nested list, and flatten promoting address.city into its own dotted top-level column instead of a nested struct. See Response formats.

Schema drift handling — adding a rescued_data_column to the schema so that any top-level keys the API returns outside the declared schema are captured as a JSON string instead of silently dropped, with a commented-out fail_on_new_fields as the alternative that raises immediately instead. See Schema.

Passing the schema through the schema option instead of .schema(...), serialized with ITEMS_SCHEMA.json() — Spark's own JSON schema representation:

.option("schema", ITEMS_SCHEMA.json())

infer_schema with schema_hints — no schema on hand at all, sampling the first page and inferring one, with schema_hints overriding individual inferred types (including a dotted path into a nested field) and a commented-out infer_column_types=false for the conservative every-leaf-a-string mode:

.option("infer_schema", "true")
.option("schema_hints", "updated_at timestamp, user.dob date")
# .option("infer_column_types", "false")

Non-JSON response formats — a csv read (header row required, typed via infer_schema) and an xml read (records_path pointing at the repeated element, items.item):

df_csv = (
    spark.read.format("rest_api")
    .option("url", "https://api.example.com/v1/report.csv")
    .option("response_format", "csv")
    .option("infer_schema", "true")
    .load()
)

df_xml = (
    spark.read.format("rest_api")
    .option("url", "https://api.example.com/v1/items.xml")
    .option("response_format", "xml")
    .option("records_path", "items.item")
    .option("infer_schema", "true")
    .load()
)

The file closes by pointing at auth_examples.py for every auth_type side by side.

auth_examples.py — every auth type, side by side

This file has no single "main" read — it's a sequence of independent, copy-pasteable blocks, one per auth_type, sharing the same small ITEMS_SCHEMA. Every secret is resolved via dbutils.secrets.get(...) on the driver before it touches an .option(...) call, exactly as in batch_example.py.

none — no auth at all:

df_none = (
    spark.read.format("rest_api")
    .schema(ITEMS_SCHEMA)
    .option("url", "https://api.example.com/v1/public-items")
    .load()
)

basic — HTTP Basic auth with username/password:

df_basic = (
    spark.read.format("rest_api")
    .schema(ITEMS_SCHEMA)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "basic")
    .option("username", username)
    .option("password", password)
    .load()
)

bearer — a static bearer token via token:

df_bearer = (
    spark.read.format("rest_api")
    .schema(ITEMS_SCHEMA)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "bearer")
    .option("token", token)
    .load()
)

api_key as a header (the default api_key_location) — api_key_value plus a custom header name via api_key_name:

df_api_key_header = (
    spark.read.format("rest_api")
    .schema(ITEMS_SCHEMA)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "api_key")
    .option("api_key_value", api_key)
    .option("api_key_name", "X-Api-Key")
    .load()
)

api_key as a query param — same auth type, api_key_location switched to query:

df_api_key_query = (
    spark.read.format("rest_api")
    .schema(ITEMS_SCHEMA)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "api_key")
    .option("api_key_value", api_key)
    .option("api_key_name", "api_key")
    .option("api_key_location", "query")
    .load()
)

header — the escape hatch for custom/non-standard schemes: an auth_headers JSON object merged into every request verbatim, useful for a vendor-specific signing header or several headers required together:

df_custom_header = (
    spark.read.format("rest_api")
    .schema(ITEMS_SCHEMA)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "header")
    .option("auth_headers", f'{{"X-Signing-Key": "{signing_key}", "X-Client-Id": "acme-corp"}}')
    .load()
)

oauth2_client_credentials — the RFC 6749 client_credentials grant, with an optional oauth_scope:

df_oauth2 = (
    spark.read.format("rest_api")
    .schema(ITEMS_SCHEMA)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "oauth2_client_credentials")
    .option("oauth_token_url", "https://auth.example.com/oauth/token")
    .option("oauth_client_id", client_id)
    .option("oauth_client_secret", client_secret)
    .option("oauth_scope", "items.read")
    .load()
)

The token is fetched lazily per executor process and cached until about 30 seconds before expiry, then refreshed automatically — no manual refresh logic needed. A trailing comment notes oauth_client_auth_mode="body" for auth servers that expect client_id/client_secret in the token request body instead of HTTP Basic auth on the token endpoint (the default is basic). Full option details are in Request & auth.

streaming_example.py — incremental cursor-driven streaming

The main body streams from a readStream source using api_key auth, page_number pagination, and a stream_cursor_field of updated_at, then writes to a Delta table on a 5-minute micro-batch trigger:

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()),
    ]
)

api_key = dbutils.secrets.get("api-scope", "api-key")

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", 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")
)

Each micro-batch pulls everything with updated_at greater than the last seen value, using whichever pagination_type is configured, and Spark handles checkpointing between batches. Full mechanics — including the pyspark >= 4.2 requirement for restart/checkpoint safety — are in Streaming.

Two commented-out alternatives follow the main example:

A numeric, non-zero-padded cursor — e.g. an auto-increment row id. The default cursor comparison is lexicographic (a string comparison), which is correct for ISO-8601 timestamps and zero-padded ids but wrong for raw numbers once they cross a digit-count boundary ("9" > "10" as strings). Setting stream_cursor_is_numeric fixes the comparison; it does not change how the cursor value itself is sent upstream:

.option("stream_cursor_field", "row_id")
.option("stream_cursor_param", "since_id")
.option("stream_initial_value", "0")
.option("stream_cursor_is_numeric", "true")

A nested cursor field — for a response like {"id": 1, "meta": {"updated_at": "..."}}, stream_cursor_field supports the same dot-path syntax as records_path/next_cursor_path:

.option("stream_cursor_field", "meta.updated_at")
.option("stream_cursor_param", "updated_since")
.option("stream_initial_value", "1970-01-01T00:00:00Z")

Where to go next

• Getting started for the quick-start walkthrough and secrets handling.
• Request & auth, Response formats, Schema, Pagination & partitioning, Streaming, and Error handling for the full option reference behind everything shown above.
• Development for running the test suite locally.