Connecting ADF to REST APIs and Web Sources

"Get the data from their API" sounds like one task. In Azure Data Factory it is four possible tools, and connecting ADF to REST APIs goes wrong most often at the first step: picking the tool meant for a different job.

The definitions below are written from Microsoft's documentation and have not been deployed.

Four tools, four jobs

Microsoft's own distinction between the first two: the REST connector is built for RESTful APIs, and the HTTP connector is generic, for example to download a file. Using HTTP against an API works but is less functional (Microsoft Learn: HTTP connector). A third connector, Web table, extracts a table from an HTML page.

The practical rule: data goes through a connector, control goes through an activity. A Web activity that fetches ten thousand records is the wrong tool, for reasons its limits make plain below.

The REST connector

What it supports (Microsoft Learn: REST connector):

• As a source, GET or POST. As a sink, POST, PUT or PATCH.
• JSON only. Only JSON response payloads are supported, and the connector sets the Accept header to application/json itself.
• Authentication: anonymous, basic, service principal, OAuth2 client credential, and system-assigned or user-assigned managed identity. An API key is handled as anonymous authentication plus an authentication header.
• Both runtimes. The Azure integration runtime for public APIs, a self-hosted one for APIs inside a private network.

A linked service holds the base URL and the authentication:

{
  "name": "OrdersApi",
  "properties": {
    "type": "RestService",
    "typeProperties": {
      "url": "https://api.example.com/",
      "enableServerCertificateValidation": true,
      "authenticationType": "OAuth2ClientCredential",
      "tokenEndpoint": "https://login.example.com/oauth2/token",
      "clientId": "00000000-0000-0000-0000-000000000000",
      "clientSecret": {
        "type": "AzureKeyVaultSecret",
        "store": { "referenceName": "CorpKeyVault", "type": "LinkedServiceReference" },
        "secretName": "orders-api-client-secret"
      },
      "scope": "orders.read"
    }
  }
}

The dataset adds a relative URL, and the Copy activity's source carries the request details:

Linked services and Key Vault integration cover the connection and the secret.

Pagination

An API rarely returns everything in one response. It returns a page, and information about how to ask for the next one. Pagination rules tell the connector how to compose each following request from the current response.

The rules are a dictionary. Keys say which part of the next request to set; values say where to find it:

A value is either a header of the current response or a JSONPath expression into its body, such as $.paging.next. There is also a range form for APIs that page by offset. A source that follows a next link and also stops when a page comes back with no data:

{
  "source": {
    "type": "RestSource",
    "requestMethod": "GET",
    "paginationRules": {
      "AbsoluteUrl": "$.paging.next",
      "EndCondition:$.data": "Empty"
    },
    "httpRequestTimeout": "00:01:40",
    "requestInterval": "00:00:01"
  }
}

When it stops. By default the connector stops when it receives HTTP 204, no content, or when a JSONPath in the rules returns null. Microsoft's documentation also warns about the case where it never stops: if the last page's next link is the same as the last request's URL, the connector sends requests endlessly, and an end condition is needed.

Three more documented details:

• With no rule defined, links in the response headers in RFC 5988 style are followed by default.
• Pagination is not supported when the top level of the response is a JSON array.
• The range form and some syntax differ between the Copy activity and mapping data flows.

Pagination behaviour, simulated

The script below starts a small local API over the 412 invoices of the Chinook sample database (Chinook on GitHub), 100 per page, and a client that follows five rules in the way the documentation describes. It is a Python simulation of the logic, not the connector, and it calls nothing outside the machine.

import itertools, json, sqlite3, threading, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

ROWS = [dict(zip(("InvoiceId", "CustomerId", "Total"), r)) for r in
        sqlite3.connect("chinook.sqlite").execute("SELECT InvoiceId, CustomerId, Total FROM Invoice ORDER BY InvoiceId")]
PAGE = 100


class Api(BaseHTTPRequestHandler):
    """A small paged API over the 412 Chinook invoices, in three common styles."""
    def do_GET(self):
        url = urlparse(self.path)
        q = parse_qs(url.query)
        here = f"http://{self.headers['Host']}{url.path}"
        if url.path == "/offset":                                   # ?offset=N; an empty list past the end
            start = int(q["offset"][0])
            body = {"data": ROWS[start:start + PAGE]}
        else:                                                       # ?page=N; a link to the next page
            page = int(q["page"][0])
            last = page * PAGE >= len(ROWS)
            body = {"data": ROWS[(page - 1) * PAGE:page * PAGE], "isLast": last, "paging": {}}
            if not last:
                body["paging"]["next"] = f"{here}?page={page + 1}"
            elif url.path == "/sticky":                             # a real-world quirk: the last page links to itself
                body["paging"]["next"] = f"{here}?page={page}"
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(body).encode())

    def log_message(self, *args):
        pass


def follow(first_url, json_path):
    """Next request's URL = a property of the previous response body, e.g. $.paging.next"""
    def next_url(previous):
        if previous is None:
            return first_url
        for key in json_path.lstrip("$.").split("."):
            previous = previous.get(key) if isinstance(previous, dict) else None
        return previous
    return next_url


def by_offset(template, offsets):
    """Next request's query parameter = the next value of a range"""
    offsets = iter(offsets)
    def next_url(previous):
        offset = next(offsets, None)
        return None if offset is None else template.format(offset=offset)
    return next_url


def copy(next_url, end=None, max_requests=None):
    rows, requests, response = [], 0, None
    while max_requests is None or requests < max_requests:
        url = next_url(response)
        if not url:                                                 # the rule produced nothing: stop
            break
        response = json.load(urllib.request.urlopen(url))
        requests += 1
        rows += response["data"]
        if end and end(response):                                   # an explicit end condition
            break
    return requests, len(rows), len({r["InvoiceId"] for r in rows})


server = ThreadingHTTPServer(("127.0.0.1", 0), Api)
threading.Thread(target=server.serve_forever, daemon=True).start()
root = f"http://127.0.0.1:{server.server_port}"

runs = {
    "next link in the body":                     copy(follow(f"{root}/pages?page=1", "$.paging.next")),
    "offset range 0 to 400 by 100":              copy(by_offset(root + "/offset?offset={offset}", range(0, 401, PAGE))),
    "open offset range, end when data is empty": copy(by_offset(root + "/offset?offset={offset}", itertools.count(0, PAGE)),
                                                      end=lambda r: r["data"] == []),
    "self-linking last page, limit 8 requests":  copy(follow(f"{root}/sticky?page=1", "$.paging.next"), max_requests=8),
    "self-linking last page, end on isLast":     copy(follow(f"{root}/sticky?page=1", "$.paging.next"),
                                                      end=lambda r: r["isLast"] is True),
}
print(f"{'pagination rule':44} {'requests':>8} {'rows copied':>12} {'distinct':>9}")
for name, (requests, rows, distinct) in runs.items():
    print(f"{name:44} {requests:8} {rows:12} {distinct:9}")
server.shutdown()

pagination rule                              requests  rows copied  distinct
next link in the body                               5          412       412
offset range 0 to 400 by 100                        5          412       412
open offset range, end when data is empty           6          412       412
self-linking last page, limit 8 requests            8          448       412
self-linking last page, end on isLast               5          412       412

What the five lines show:

• A next link in the body takes 5 requests for 412 rows and stops when the last page has no link.
• A fixed offset range also takes 5 requests, and depends on your knowing the upper limit in advance. When the table grows past it, rows are silently missed.
• An open range with an end condition takes 6. The sixth request returns an empty page, which is how the client learns it has finished. One extra call is the price of not hard-coding the size.
• A last page that links to itself, stopped only by a request limit, makes 8 requests and copies 448 rows, of which 412 are distinct. The last page was loaded four times. A maximum request number prevents an endless run; it does not prevent duplicates.
• The same API with an end condition on its "is last" flag is back to 5 requests and 412 rows.

The lesson holds for any paged copy: read the API's last page by hand before trusting a rule, and make the load safe against duplicates, by landing raw pages and deduplicating on the key, or by upserting.

From JSON to rows

There are two documented ways to handle the response:

• Export as-is. With no mapping defined, the JSON is written unchanged to a file store.
• Schema mapping. Fields are mapped to columns by JSON path, with one array chosen as the collection reference to produce a row per element (Microsoft Learn: schema mapping).

For anything beyond a flat response, landing the raw JSON in the lake first and shaping it afterwards is the more forgiving design. The original responses are kept, and a changed field in the API does not break the ingestion.

The Web activity

The Web activity calls an endpoint and returns the response to the pipeline, where later activities read it (Microsoft Learn: Web activity). Its limits define its role:

• The response must be JSON, and the maximum response size is 4 MB.
• Timeout. One minute by default; it can be raised to ten minutes.
• Methods: GET, POST, PUT, PATCH and DELETE.
• Long-running calls. If the endpoint follows the asynchronous request-reply pattern, returning 202 with a location to poll, the activity keeps polling.

Use it to start a job in another system, request a token, post a notification or read a small configuration document. Do not use it to move data.

The Webhook activity

A Webhook activity posts to an endpoint and then waits. The service adds a callBackUri to the request body, and the pipeline continues only when the other system invokes that URI (Microsoft Learn: Webhook activity). If it is not invoked within the timeout, 10 minutes by default, the activity fails as timed out. With the report status on callback option, the caller can pass back a failure and an error message.

This is the tool for "run this external process and continue when it finishes" without polling.

Practical concerns

• Throughput. A copy from REST uses 1 Data Integration Unit (Microsoft Learn: performance features). Pages are fetched one after another. To go faster, split the work by something the API can filter on, such as date ranges, and run several copies from a ForEach, within the API's rate limits.
• Rate limits. The request interval is the built-in throttle. Set it from the API's published limits.
• Incremental pulls. Pass a "modified since" value as a pipeline parameter into the relative URL or query, and store the high-water mark. Parameterizing pipelines covers the mechanism.
• Failures midway. A failed run that restarts from page one will refetch pages. Write each run to its own folder and replace, not append. Handling errors and retries covers retry settings.

For how such APIs are built, which helps when reading their documentation, see building REST APIs with Python. The activity that carries the connector is covered in the Copy activity guide.

Azure Data Factory and API-based data ingestion are covered in the Data Engineering course.

Sources

• Copy and transform data from and to a REST endpoint (Microsoft Learn) — Microsoft Learn
• Copy data from an HTTP source (Microsoft Learn) — Microsoft Learn
• Web Activity (Microsoft Learn) — Microsoft Learn
• Webhook activity (Microsoft Learn) — Microsoft Learn
• Schema and data type mapping in copy activity (Microsoft Learn) — Microsoft Learn
• Copy activity performance optimization features (Microsoft Learn) — Microsoft Learn
• Chinook sample database (GitHub) — GitHub