spark-rest-api-reader

Full documentation: https://liam-mcelhaney122.github.io/spark-rest-api-reader/

A generic Spark Python Data Source for pulling data out of any GET REST API — returning JSON, CSV, or XML — and into a DataFrame, batch or streaming, without writing a bespoke connector for every API.

Requires Spark 4.0+ / Databricks Runtime 15.4+ (the Python Data Source API). Not tied to Databricks otherwise — it works with any Spark 4.0+ cluster.

Why

Every project ends up writing the same HTTP-get + pagination + retry + auth-header boilerplate to land a REST API into Bronze. This wraps that boilerplate once, behind Spark options, so pulling a new API is a matter of configuring auth_type/pagination_type, not writing new Python.

Built on httpx for the HTTP client and tenacity for retry/backoff.

Install

pip install spark-rest-api-reader

On a Databricks cluster, install it as a cluster library (PyPI source: spark-rest-api-reader) or per-notebook:

%pip install spark-rest-api-reader

For air-gapped clusters, build the wheel yourself and upload it as a cluster library:

pip install build
python -m build   # produces dist/spark_rest_api_reader-*.whl

pyspark/pyarrow are intentionally not pinned as install dependencies — the cluster already provides them, and pinning risks conflicting with the runtime's own version. For local development/tests, install .[dev].

Quick start

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

from spark_rest_api_reader import RestApiDataSource

spark.dataSource.register(RestApiDataSource)

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

df = (
    spark.read.format("rest_api")
    .schema(items_schema)
    .option("url", "https://api.example.com/v1/items")
    .option("auth_type", "bearer")
    .option("token", dbutils.secrets.get("api-scope", "api-token"))
    .option("pagination_type", "page_number")
    .load()
)

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

spark.dataSource.register only needs to run once per session (e.g. once at the top of a job/notebook).

Secrets: resolve dbutils.secrets.get(...) (or any other secret lookup) on the driver before passing the value into .option(...). Options are plain strings that get shipped to executors as part of the reader — do not put a dbutils call anywhere reachable from executor code, since dbutils isn't available there.

How it's organized

• Auth (auth_type) — how the request is authenticated.
• Pagination (pagination_type) — how a single partition walks through multiple pages of results until the API says "no more."
• Partitioning (partitions / pages) — how the overall pull is split into independent units of work that Spark can fan out across executors.
• Response format (response_format) — how the response body is parsed: json (default), csv, or xml. Everything downstream (records extraction, schemas, pagination stop-conditions) works the same on all three.
• Records extraction (records_path, flatten) — how to find the list of records inside the parsed response body, and whether to flatten nested objects into dotted column names.
• Schema — supply a StructType via .schema(...) (or the schema option), or set infer_schema=true to sample and infer one, Auto Loader-style.
• Streaming — an incremental variant driven by a cursor field in the response (e.g. updated_at), via spark.readStream.

Partitioning and pagination are independent: e.g. you can partition by region (partitions) and have each region's partition separately paginate through everything on its own (pagination_type), or auto-generate one partition per page number (pages) for maximum parallelism when the API doesn't need per-partition pagination at all.

Option reference

Request basics

Auth (auth_type)

OAuth2 tokens are fetched lazily per executor process and cached until ~30 seconds before expiry, then refreshed automatically.

Response format

• csv — a header row is required; each data row becomes one record, every value a string (coerce via the schema, or let infer_schema sniff the types). Empty cells become null. Incompatible with records_path (the body already is the record list) and with pagination_type=cursor (no envelope to read a next-cursor from).
• xml — the element tree maps onto nested dicts/lists: repeated elements become arrays, attributes become {prefix}-prefixed keys, and an element with both attributes and text keeps its text under {prefix}VALUE. Dot paths (records_path, next_cursor_path, stream_cursor_field) and nested-struct schemas then work exactly as for JSON; paths are relative to inside the root element (e.g. items.item for <response><items><item>...).

Records extraction

Schema

Schemas follow the structs pattern: nested objects are declared and returned as real StructType/ArrayType columns — build them as StructType objects, not DDL strings, and never model nested data as JSON-string columns.

Manually defined (preferred for production jobs):

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

schema = StructType(
    [
        StructField("id", IntegerType()),
        StructField("address", StructType([StructField("city", StringType())])),
        StructField("tags", ArrayType(StringType())),
    ]
)

df = spark.read.format("rest_api").schema(schema).option("url", ...).load()

Inferred (Auto Loader-style, for all three response formats):

Inference merges field sets across sampled records, widens mixed numeric samples to double (and mixed date/timestamp to timestamp), and falls back to string only on genuine type conflicts.

Nested response objects map onto nested StructType/ArrayType schema fields automatically. ISO-8601 timestamp/date strings are parsed for TimestampType/DateType columns.

Type coercion: values are best-effort coerced to match the declared column type — a stringified number/boolean ("123", "true") coerces to Integer/Boolean, and a non-string value in a StringType column is stringified. This matters because REST APIs commonly stringify numbers to dodge JS float-precision issues. A value that genuinely can't be coerced (e.g. an object where a scalar is expected) raises SchemaTypeError naming the exact field path and both types involved, instead of silently reaching Spark's Arrow conversion and failing there with an opaque, unattributed error several layers removed from the actual problem.

Schema drift

The schema is fixed once resolved — it is not re-inferred mid-batch or mid-stream. By default, fields the API returns that aren't in the schema are silently dropped (and fields the schema expects but the API omits become null). Two options change that:

Drift detection is top-level only — new keys appearing inside an array element aren't (currently) surfaced by either option. One nuance: with flatten=true, nested dict keys are promoted to top-level dotted names before drift detection runs, so nested-dict drift actually is caught in that combination — only drift nested inside an array stays invisible either way.

In-band errors

Some APIs report failure inside a successful HTTP response — ArcGIS, for example, answers 200 OK with {"error": {...}} in the body. raise_for_status() never sees those; untreated, they surface as confusing downstream symptoms ("zero records to infer from" at schema inference, or pagination silently ending early and truncating the data). Two composable mechanisms catch them; every page of every request is checked, including the schema-inference sample:

For checks a dot path can't express, attach a response_validator function to a RestApiDataSource subclass and register the subclass under its own name. The validator receives the parsed body of each page and returns an error message to fail the read (or None to accept it); raising its own exception also fails the read. When both are configured, error_path is checked first.

from spark_rest_api_reader import RestApiDataSource

class ArcGisDataSource(RestApiDataSource):
    @classmethod
    def name(cls):
        return "arcgis_rest"

    @staticmethod
    def response_validator(body):
        if isinstance(body, dict) and "error" in body:
            return f"ArcGIS error: {body['error']}"

spark.dataSource.register(ArcGisDataSource)

df = spark.read.format("arcgis_rest").option("url", ...).load()

It must be a subclass attribute rather than an .option() because option values are plain strings, and because data source code runs in isolated worker processes: a subclass defined in a notebook is cloudpickled by value at register() time, so the attached function travels to every worker — module-level state (a registry, a global) would not.

Pagination (pagination_type) — within one partition

max_pages_per_partition (default 10000) is a safety valve against runaway pagination loops.

Partitioning — how the pull fans out across Spark

Streaming (spark.readStream)

Each micro-batch pulls everything from stream_cursor_field > last value using whatever pagination_type is configured, all within a single partition. Simple and correct; if you need intra-batch parallelism, prefer the batch reader with partitions on a schedule instead.

Cursor comparison: by default, "greater than" is a string comparison — correct for ISO-8601 timestamps and zero-padded numeric IDs, but wrong for raw (non-zero-padded) numeric cursors once they cross a digit-count boundary ("9" > "10" is True lexicographically). Set stream_cursor_is_numeric=true for a numeric cursor field to compare correctly instead.

Restart/checkpoint safety requires pyspark >= 4.2. This reader implements the admission-control latestOffset(start, limit) signature, which is the only one that actually receives the restored checkpoint offset after a query restart (driver restart, job retry, redeploy, etc.) — pyspark added this in 4.2.0. On an older pyspark/DBR where only the legacy zero-arg latestOffset() convention exists, the reader still works, but the cursor can only be tracked in an in-memory attribute that resets on every fresh process: a restart resumes from stream_initial_value again, reprocessing everything already ingested, rather than crashing. 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.

Examples

• examples/batch_example.py — batch read into Delta, with commented alternatives for every pagination style (page_number, offset_limit, cursor, link_header), auto page-range/explicit-partition fan-out, records_path/flatten for custom envelopes, schema drift handling (rescued_data_column/fail_on_new_fields), passing a StructType through the schema option, infer_schema with schema_hints, and csv/xml response formats.
• examples/auth_examples.py — every auth_type (none, basic, bearer, api_key as a header or query param, header for custom schemes, oauth2_client_credentials) as an independent, copy-pasteable snippet.
• examples/streaming_example.py — incremental streaming pull driven by a cursor field, with Spark handling checkpointing between micro-batches.

Development

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest

Tests spin up a real local SparkSession plus a real local HTTP server (not mocks) so that requests issued from PySpark's separate worker processes are exercised end-to-end.

See CONTRIBUTING.md for the full development guide, and CHANGELOG.md for release history.

License

Apache-2.0