Extraction: Pulling Data Reliably

What you will learn

1. The three promises every extractor makes.
2. Pages that do not skip or repeat rows when new rows arrive.
3. A watermark that survives a tie at the boundary.
4. Rate limits, Retry-After, and backoff with jitter.
5. Which errors to retry, and how to refresh a token.
6. How to split a big read into safe parts.

Extraction is the step that reads data out of a source and into your pipeline. You met it in Data Sources & Ingestion Patterns (chapter 05). This chapter goes deeper.

Here is a quick recap of the words you need. A watermark is the saved value that marks how far the last run read. A cursor is a bookmark the API gives you.

Status code 429 means too many requests. A header is an extra line of information that comes with an answer. A Retry-After header says how long to wait.

From Change Data Capture (CDC) (chapter 06) you know that a timestamp watermark misses deletes. This chapter keeps the watermark and makes it safe. For deletes, use CDC.

One fact explains the whole chapter. If extraction misses a row, the next steps never know that the row existed. So extraction must be right.

Every extractor makes three promises. If it breaks one, the data is wrong, and often nobody sees it.

1. Complete: every row that should arrive does arrive.
2. Safe repeats: a row that arrives twice counts only once.
3. Resumable: after a crash, the job continues where it stopped.

The three work together. To resume, the job re-reads a little, and that can repeat rows. So repeats must be harmless.

A checkpoint is a saved position, such as a cursor or a watermark. The job saves it after each page it has written. This is called at-least-once: no row is lost, and a few may repeat.

Idempotency, Retries & Exactly-Once (chapter 11) covers the safe write in full. Here, watch the checkpoint work. Each page holds three orders, and the cursor is the last order_id read.

cursor = load_checkpoint()          # None on run 1, "103" on run 2
while True:
    page = fetch_page(after=cursor, limit=3)
    write_to_landing(page["data"])  # write first
    cursor = page["next_cursor"]
    if cursor is None:              # the source says: no more pages
        break
    save_checkpoint(cursor)         # save second

A write by key means update or insert by order_id. Chapter 05 did this in its orders load.

On the last page there is no new checkpoint, so every later run re-reads that page. The write by key makes that harmless.

An API returns rows in pages. A page is a small group of rows, such as two payments. You ask again until the API says it is done.

There are two ways to ask for the next page. Offset says "skip N rows, then give me M rows". A keyset cursor says "give me the rows after this key".

The API calls in this chapter start with GET. GET is a web call that only asks for data.

Offset counts rows, so it breaks when rows change while you read. Here the job asks for the newest payment first, two per page. Payment p_9004 is made up and arrives between call 1 and call 2.

GET /payments?order=newest&offset=0&limit=2   # call 1
GET /payments?order=newest&offset=2&limit=2   # call 2
GET /payments?order=newest&offset=4&limit=2   # call 3

Deletes cause the opposite. Say instead that no payment arrives, and p_9003 is deleted after call 1. The list is now p_9002, p_9001, so call 2 (offset 2) gets nothing and p_9001 is never read.

Deep offsets are also usually slower on big tables. The database must walk past every skipped row.

A keyset cursor does not count rows. It remembers the last key, and the next call asks for keys greater than it. This works because each new payment gets a larger payment_id.

In chapter 05 the API gave a cursor you cannot read, such as c2. To show the idea, this cursor is the plain payment_id.

Here are the same payments, oldest first. Again p_9004 arrives between call 1 and call 2.

GET /payments?limit=2                  # call 1
GET /payments?limit=2&after=p_9002     # call 2
GET /payments?limit=2&after=p_9004     # call 3

Pass a real cursor back unchanged. Do not build or edit it yourself.

If a payment with a smaller id saves late, the keyset can still miss it. The next section shows the same problem with timestamps.

The key must be unique. If you sort by updated_at, many rows can share one value. Sort by updated_at, then by payment_id, so the cursor points to one place.

An incremental load reads only rows that are new or changed. Its watermark is the largest value of one column that you have loaded. That column is the incremental key.

Pick a key that changes whenever the row changes. Here are three choices for the orders table.

updated_at is the usual choice. A bulk fix or a bug can still change rows without moving it. The watermark never sees those rows.

Now a trap: ties. Many rows can carry the same updated_at. A transaction is a group of changes that are saved together.

Here the app sets updated_at with now(). In PostgreSQL, now() gives the time the transaction started, not the time it ended. Other readers see the row only when the transaction ends.

Orders 105 and 189 share the stamp 09:00:00 on 2026-01-25. Order 189 is made up.

The stamps are rounded to seconds. Follow the timeline.

First, run B with a strict > filter.

SELECT order_id, customer_id, status, amount, updated_at
FROM   orders
WHERE  updated_at > '2026-01-25 09:00:00'   -- the saved watermark

Now run B with >=. It reads the boundary row again, so order 105 comes twice.

SELECT order_id, customer_id, status, amount, updated_at
FROM   orders
WHERE  updated_at >= '2026-01-25 09:00:00'
ORDER  BY order_id;

With a write by key, run B only updates order 105. Some landing tables only add rows, and then the repeat stays.

Then remove it when you read. DISTINCT keeps one copy of rows that match in every column you select.

SELECT DISTINCT order_id, status, amount, updated_at
FROM   landing_orders
ORDER  BY order_id;

The >= filter fixes only a tie exactly at the watermark. A row whose stamp is older than the watermark, but that saves after the read, is still missed. A lookback re-reads a few days to catch it.

Incremental Processing & Backfills (chapter 12) covers the lookback and the half-open window.

A rate limit caps how many calls you may make in a period. The API checks it. When you go over, it answers 429.

The answer may carry a Retry-After header. It holds a number of seconds, or a date. Both mean "wait this long".

This code turns either form into seconds. The date form is an HTTP date, in UTC (the world standard time). It ends with the letters GMT.

from datetime import datetime, timezone
from email.utils import parsedate_to_datetime

def retry_after_seconds(value):
    if value.isdigit():                      # "7"
        return int(value)
    when = parsedate_to_datetime(value)      # "Mon, 26 Jan 2026 09:00:07 GMT"
    return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())

What if there is no Retry-After? Do not retry at once. Many fast retries make the problem worse.

Exponential backoff means each wait limit is larger than the last. Here the limit doubles up to a cap. Jitter means you pick a random time inside the limit.

Jitter matters when many workers fail together. A worker is one copy of the job that runs at the same time as others.

Without jitter, all workers retry at the same moments and hit the API together again. The AWS Architecture Blog post "Exponential Backoff And Jitter" describes this and recommends jitter.

import random

def backoff_seconds(attempt, base=1.0, cap=10.0):
    limit = min(cap, base * 2 ** attempt)    # 1, 2, 4, 8, 10, 10 ...
    return random.uniform(0, limit)          # a random time up to the limit

Now put both rules in one loop. A Retry-After header wins when it exists. The code gives up after 5 tries.

import time

def get_with_backoff(session, url, params, max_tries=5):
    for attempt in range(max_tries):
        resp = session.get(url, params=params, timeout=(5, 30))
        if resp.status_code != 429:
            resp.raise_for_status()
            return resp
        header = resp.headers.get("Retry-After")
        wait = retry_after_seconds(header) if header else backoff_seconds(attempt)
        if attempt == max_tries - 1:
            break
        time.sleep(wait)
    raise RuntimeError(f"still rate limited after {max_tries} tries")

Not every error deserves a retry. A retry helps when the problem is short. It does not help when the request itself is wrong.

A call that only reads, such as GET, is safe to repeat. This table says what the job does for each answer.

Many APIs use OAuth. OAuth is a common standard for giving a program access to an API. Your job sends an access token with each call.

A token is a secret string that proves who you are. It expires after a short time, and each API chooses how long. Bearer is the word OAuth puts before the token in the header.

The job also holds a longer-lived secret, such as a refresh token. It trades that for a new access token. The details differ by API, so read its documents.

A long run can outlive its token. Here the token expires at page 41 of 90. The job refreshes it and keeps its cursor.

def get_page(session, url, cursor, token):
    headers = {"Authorization": f"Bearer {token}"}
    resp = session.get(url, params={"after": cursor}, headers=headers, timeout=(5, 30))
    if resp.status_code == 401:
        token = refresh_access_token()     # made-up helper
        headers = {"Authorization": f"Bearer {token}"}
        resp = session.get(url, params={"after": cursor}, headers=headers, timeout=(5, 30))
    resp.raise_for_status()                # a second 401 stops the job
    return resp.json(), token

One worker reading a huge table may not finish in time. Parallel extraction splits the table into parts. Each part goes to its own worker, and the parts run at the same time.

A partition is one such part. Cut by a key that never changes for a row, such as order_date or order_id. Use half-open ranges, which include the start and exclude the end.

Incremental Processing & Backfills (chapter 12) uses the same idea for time windows.

Here are the six orders, cut into three date ranges. One worker reads each range.

-- worker 1
SELECT order_id, order_date FROM orders
WHERE  order_date >= '2026-01-01' AND order_date < '2026-01-11';
-- worker 2
SELECT order_id, order_date FROM orders
WHERE  order_date >= '2026-01-11' AND order_date < '2026-01-21';
-- worker 3
SELECT order_id, order_date FROM orders
WHERE  order_date >= '2026-01-21' AND order_date < '2026-02-01';

Now the same split with BETWEEN, which includes both ends. Orders on the boundary days fall into two parts.

-- worker 1: BETWEEN '2026-01-01' AND '2026-01-11'
-- worker 2: BETWEEN '2026-01-11' AND '2026-01-21'
-- worker 3: BETWEEN '2026-01-21' AND '2026-01-31'

Parts also help recovery. Each part has its own checkpoint. If part 2 fails, you run only part 2 again, which reads 3 orders.

Replace the whole part when you load it. Then a half-written part leaves no repeats. Idempotency, Retries & Exactly-Once (chapter 11) covers this.

Parts can be uneven. A busy day can hold far more rows than a quiet day, and an id range with gaps can hold few rows.

A laptop version works on a small test. A production version survives big data and failures. Each extractor below has both versions.

Extractor 1: the database

In a database, a cursor is a different thing from an API cursor. It is a query result that stays on the database server, and you fetch it a few rows at a time.

One big list works for six orders. It fails on millions, because memory fills.

psycopg2 is a Python library for PostgreSQL. A library is ready-made code you import. A batch is a small group of rows, such as two orders.

A named cursor keeps the result on the database server, and you fetch one batch at a time.

import psycopg2

conn = psycopg2.connect(dsn)                # dsn holds the connection details
with conn.cursor(name="orders_read") as cur:   # a name makes it a database cursor
    cur.execute("SELECT order_id, customer_id, status, amount FROM orders ORDER BY order_id")
    while True:
        batch = cur.fetchmany(2)            # use thousands in real jobs
        if not batch:
            break
        write_to_landing(batch)             # made-up helper

Extractor 2: the API

requests is a Python library for web calls. It has no default timeout, so a stuck call can wait for ever. The scheduler still shows the job as running.

Set two timeouts: one to connect, one to read.

import requests

session = requests.Session()     # reuses the connection between pages
resp = session.get(PAYMENTS_URL, params={"limit": 100}, timeout=(5, 30))
# 5 seconds to connect, 30 seconds to wait for data

Extractor 3: no updated_at

Some tables have no updated_at. Load a full copy into a staging table, then compare it with the warehouse. A staging table holds today’s full copy for a short time.

IS DISTINCT FROM treats NULL as a value.

SELECT t.order_id, t.status, t.amount
FROM   staging_orders t
LEFT   JOIN dw_orders w ON w.order_id = t.order_id
WHERE  w.order_id IS NULL                      -- new row
   OR  w.status IS DISTINCT FROM t.status      -- changed
   OR  w.amount IS DISTINCT FROM t.amount      -- changed
ORDER  BY t.order_id;

Extractor 4: passwords and secrets

A password in code stays in the history of your code (git), and anyone who reads the code sees it. A secret store is a safe place that keeps passwords. Fetch them from there when the job runs.

boto3 is the Python library for AWS (Amazon Web Services). The secret name here is made up.

import json
import boto3

def get_api_key(secret_id):
    client = boto3.client("secretsmanager")
    value = client.get_secret_value(SecretId=secret_id)["SecretString"]
    return json.loads(value)["api_key"]

api_key = get_api_key("shop/payments-api")   # made-up name

• Extraction must be complete, safe to repeat, and resumable. Write first, then save the checkpoint.
• Offset paging breaks when rows change. A keyset cursor does not skip or repeat rows when rows are added or removed, except rows that save late. Stop when the API says stop.
• With a tie at the watermark, > loses rows. Use >= and remove the repeats.
• Wait for Retry-After, or use backoff with jitter. Cap the tries. Retry only short errors, such as 429 and 503.
• Cut big reads into half-open parts. Retry one part at a time, and ask the owner how many workers you may use.

Five scenarios. Build each one on the pipeline canvas with the chapter closed. The review tells you which requirement your design missed.