Ingest data from an API in pipelines

Ingesting from an API means pulling data over HTTP from a web service, usually as paginated JSON, rather than reading from a file or a database. Unlike files or a message bus, there's no built-in generic API source, so you handle authentication, pagination, and rate limits yourself. Lakeflow pipelines support three patterns for ingesting from an arbitrary API. Which one fits depends on your volume and refresh needs.

Important

Before you write any custom API-ingestion code, check whether a managed connector already exists for your source. Lakeflow Connect ships built-in connectors for many common software as a service (SaaS) APIs, such as Salesforce, Workday, ServiceNow, and Google Analytics, and there's a growing set of partner connectors as well. If a connector covers your source, it handles authentication, pagination, and incremental extraction for you, and it's almost always less work than a hand-rolled ingestion. See Managed connectors in Lakeflow Connect. Use the patterns below only when no connector fits.

Prerequisites

  • A pipeline. To create one, see Lakeflow pipelines tutorials.
  • API credentials, such as a token or key, stored as a Azure Databricks secret. Never hardcode credentials in pipeline source code. See Secret management.
  • Network access from your pipeline compute to the API endpoint.
  • Familiarity with streaming tables and materialized views, the dataset types these patterns produce. See Streaming tables and Materialized views.

Choose a pattern

There's no native generic REST-API source in pipelines, so when you pull from an arbitrary API, pick one of three patterns based on data volume and how often you ingest:

Pattern Use when
Periodic pulls as a materialized view Payloads are small to medium and pulled once per pipeline run, such as reference data, daily FX rates, or a paginated but boundable API.
Python Data Source API You need to poll a high-volume or streaming API incrementally, with checkpointed progress so a restart doesn't re-read everything.
Decoupled ingestion with Auto Loader You want to isolate API-specific quirks from your transformation logic and get exactly-once file tracking for free.

Pattern 1: Periodic pulls as a materialized view

For small-to-medium payloads pulled once per pipeline run, write a Python function that calls the API and returns a Spark DataFrame. Because the dataset is a materialized view, the pipeline re-runs the function fully and idempotently every time the pipeline updates.

The following steps show you how to build a materialized view with periodic pulls:

  1. Store the API token in a secret, then map it to a Spark configuration property in your pipeline settings so the pipeline code can read it. Add the property to the spark_conf block of the pipeline's cluster configuration:

    {
  "clusters": [
    {
      "spark_conf": {
        "api.token": "{{secrets/<scope-name>/<secret-name>}}"
      }
    }
  ]
}

    The code in the next step reads this value with spark.conf.get("api.token"). For more about configuring secrets in pipeline settings, see Securely access storage credentials with secrets in a pipeline.

  2. Define a materialized view that calls the API and returns the response as a DataFrame:

    import requests
from pyspark import pipelines as dp
from pyspark.sql import Row

@dp.materialized_view(
    name="exchange_rates_bronze",
    comment="Daily FX rates pulled from a public REST API",
)
def exchange_rates_bronze():
    resp = requests.get(
        "https://api.example.com/v1/rates",
        params={"base": "USD"},
        headers={"Authorization": f"Bearer {spark.conf.get('api.token')}"},
        timeout=30,
    )
    resp.raise_for_status()
    rates = resp.json()["rates"]
    rows = [Row(currency=k, rate=float(v), as_of_date=resp.json()["date"]) for k, v in rates.items()]
    return spark.createDataFrame(rows)

  3. Handle pagination inside the function by looping over pages and concatenating the results before you return the DataFrame:

    import requests
from pyspark import pipelines as dp
from pyspark.sql import Row

@dp.materialized_view(
    name="customers_bronze",
    comment="Customers pulled from a paginated REST API",
)
def customers_bronze():
    token = spark.conf.get("api.token")
    rows = []
    url = "https://api.example.com/v1/customers"
    while url:  # follow the API's next-page cursor until exhausted
        resp = requests.get(
            url,
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
        )
        resp.raise_for_status()
        payload = resp.json()
        rows.extend(Row(**record) for record in payload["data"])
        url = payload.get("next")  # None on the last page
    return spark.createDataFrame(rows)

    Add retry and backoff logic around the request for resilience.

This pattern re-reads the full API response on every pipeline update, so use it only when the payload is bounded. For incremental reads, use pattern 2.

Pattern 2: High-volume or streaming APIs with the Python Data Source API

For APIs you need to poll incrementally with offset tracking, implement a custom data source using Spark's Python Data Source API. This gives you proper streaming semantics, including checkpointed progress and incremental reads, so a restart resumes from the last offset instead of pulling the whole API again.

The following steps show you how to ingest from a custom data source:

  1. Implement a DataSource and DataSourceStreamReader that call the API and track the read offset. For details on authoring a custom data source, see PySpark custom data sources.

  2. Register the data source so the pipeline can reference it by format name:

    spark.dataSource.register(MyApiDataSource)

  3. Read from the registered source in a streaming table:

    from pyspark import pipelines as dp

@dp.table(name="events_bronze")
def events_bronze():
    return spark.readStream.format("my_api_source").load()

Pattern 3: Decouple ingestion with a scheduled job and Auto Loader

A common production pattern is to separate the API call from the pipeline. A scheduled job lands the raw API responses as files in a Unity Catalog volume, and the pipeline picks them up with Auto Loader. This isolates API-specific quirks like pagination and rate limits from your declarative transformation logic, and it gives you Auto Loader's exactly-once file tracking for free.

The following steps show you how to decouple ingestion with a scheduled job:

  1. Write a notebook or script that calls the API and writes the raw JSON responses to a Unity Catalog volume. Read the API credentials from a secret. See Secret management.

    import requests, json, time

token = dbutils.secrets.get(scope="<scope-name>", key="<secret-name>")
volume_path = "/Volumes/main/raw/landing/api_events"

resp = requests.get(
    "https://api.example.com/v1/events",
    headers={"Authorization": f"Bearer {token}"},
    timeout=30,
)
resp.raise_for_status()
# One file per run; the pipeline's Auto Loader tracks which files it has ingested.
with open(f"{volume_path}/events_{int(time.time())}.json", "w") as f:
    json.dump(resp.json()["data"], f)

  2. Schedule the notebook or script to run on its own with Lakeflow Jobs. See Lakeflow Jobs.

  3. In your pipeline, define a streaming table that reads the landed files with Auto Loader:

    from pyspark import pipelines as dp

@dp.table(name="api_events_bronze")
def api_events_bronze():
    return (
        spark.readStream.format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("/Volumes/main/raw/landing/api_events")
    )

For more about reliable file ingestion with Auto Loader, see Load files from cloud object storage and What is Auto Loader?.

Best practices for API ingestion

  • Keep secrets out of source code. Store API tokens and keys in Azure Databricks secret scopes and read them at runtime. See Secret management.
  • Validate responses early. Add expectations on the ingested rows to catch malformed API responses before they flow downstream.
  • Handle pagination and rate limits. Loop over pages and add retry with backoff so a transient failure doesn't fail the whole update.

Additional resources

  • Last updated on