Data Ingestion and Source System Integration Questions

Direct answer

Match the integration pattern to what each source can actually do, not to a single company-wide standard: the event-emitting source becomes a subscriber, the paginated-read-only source becomes a scheduled poll, and the nightly-file-export source becomes a scheduled file pickup. The design stays maintainable as requirements evolve by landing all three into the same normalized raw layer with a consistent schema and metadata, so a later change (adding near-real-time alerts from one source) only touches that one source's ingestion path, not the shared downstream model.

Structured elaboration

Matching pattern to source capability

• The event-emitting source: subscribe to its events (a webhook or a stream), land each event as it arrives; this is the only one of the three sources that can support the future near-real-time alert requirement without a redesign.
• The paginated-read-only source: a scheduled poll walking every page since the last checkpoint, on whatever interval balances freshness against the source's rate limit.
• The nightly-file-export source: a scheduled pickup job (SFTP, meaning SSH File Transfer Protocol, cloud storage, or however the export is delivered) that runs shortly after the export is expected, with a check that the file has actually landed before processing it.

Keeping the design maintainable

• Land all three into a common raw/landing layer with a consistent envelope (source name, ingestion timestamp, and the original payload), so downstream consumers work against one shape regardless of which pattern produced the data.
• Keep each source's connector isolated: a change to how the file-export source is polled should never require touching the event-subscriber's code.
• Make the freshness characteristics of each source visible downstream (a "last updated" or "as of" indicator per source) rather than presenting all three as if they were equally fresh, since silently blending a nightly-batch source with a real-time one can mislead a dashboard's audience about how current the numbers actually are.

Building in room for the requirement to evolve

• Today's ask is a daily dashboard, which the paginated and file-export sources already satisfy on their existing schedules; the event source's daily aggregation is just a rollup of its already-landed real-time stream, so no rework is needed there either.
• The stated future need (near-real-time alerts from one source) should specifically inform WHICH source you architect for streaming from day one, even if you are not yet building the alerting logic; retrofitting an already-batch-oriented connector into a streaming one later is a much larger project than building it as a subscriber from the start when you already know that is where it is headed.

Worked example

The event-driven source publishes order-status-change events; land each event onto a raw topic or table immediately as it arrives, since this is the source flagged for a future near-real-time alert. The paginated-read-only source (a support-ticketing API) is polled every 30 minutes using cursor-based pagination and a persisted checkpoint, comfortably supporting the daily dashboard's freshness need without over-polling a source that has no true real-time signal to offer anyway. The nightly-file-export source (a partner's inventory feed) is picked up by a job scheduled 30 minutes after the partner's documented export window, with a presence check before processing so a late or missing file triggers an alert rather than the job silently processing yesterday's stale file again. All three land into the same raw schema (source, ingested_at, payload), and the daily dashboard is built from a view over all three, each contributing at its own natural freshness. When the near-real-time alert requirement actually arrives, only the event source's already-real-time data needs a new consumer built against it: neither the poll-based nor the file-based ingestion paths need to change at all.

Trade-offs & pitfalls

• The most common mistake is trying to force all three sources into ONE integration pattern for "consistency," typically by wrapping the event source in a batch poll of its own API instead of subscribing directly, which throws away the one capability (real-time delivery) that source actually offers and that the stated future requirement will need.
• A file-pickup job with no presence check is a classic silent-staleness bug: if the partner's export is late or fails, a naive job that just re-reads "the file at this path" will happily reprocess yesterday's data with no error at all.
• Landing all three sources into a shared raw schema is good for maintainability, but resist the temptation to also force them into the SAME refresh cadence in that shared layer; an artificial "we refresh everything hourly" schedule either wastes effort re-polling a nightly file source or under-serves the event source's real freshness.
• Do not let "keep it maintainable" become an excuse to over-engineer a generic pluggable-connector framework for just three sources; isolating each connector's code is enough, a fully abstracted framework is a cost worth paying only once you have many more sources than three.

Direct answer

The core structure is a loop that persists its checkpoint only after each page is safely processed, retries transient failures (including rate limits) with exponential backoff, and stops rather than mis-loads when the response shape does not match what it expects. Below is a runnable implementation, executed against a deterministic fake API standing in for the real REST endpoint so the behavior shown is reproducible, not narrated.

Structured elaboration

import time
import random

random.seed(7)

EXPECTED_SCHEMA = {"id", "updated_at", "amount"}


class RateLimited(Exception):
    pass


class TransientError(Exception):
    # A 5xx or a transient network failure (timeout, connection reset); safe
    # to retry with backoff, unlike a 4xx (malformed request), which should
    # propagate immediately instead of being retried into a guaranteed failure.
    pass


def extract_incremental(get_page, start_cursor=None, max_retries=5, schema="v1"):
    # Pulls every page from `get_page` starting at `start_cursor`, persisting the
    # cursor after each successful page so a crash mid-run resumes without
    # re-pulling completed pages or losing progress. Detects a schema change by
    # comparing observed keys against EXPECTED_SCHEMA and stops (flags for a
    # human) rather than silently loading mismatched rows.
    cursor = start_cursor
    all_records = []
    checkpoints = []
    schema_alert = None

    while True:
        attempt = 0
        while True:
            try:
                page, next_cursor = get_page(cursor, schema=schema)
                break
            except (RateLimited, TransientError):
                attempt += 1
                if attempt > max_retries:
                    raise RuntimeError(f"exceeded {max_retries} retries at cursor={cursor}")
                backoff = min(2 ** attempt, 30) + random.uniform(0, 0.1)
                time.sleep(backoff)  # real backoff; skipped in the test harness below
                continue

        if page:
            observed_keys = set(page[0].keys())
            if observed_keys != EXPECTED_SCHEMA:
                schema_alert = {
                    "cursor": cursor,
                    "expected": sorted(EXPECTED_SCHEMA),
                    "observed": sorted(observed_keys),
                }
                # Stop pulling under the OLD assumption; the checkpoint already
                # reflects everything ingested before this page, so a resume
                # after the schema is fixed starts exactly here.
                break

        all_records.extend(page)
        cursor = next_cursor
        # Checkpoint is written to durable storage AFTER the page is fully
        # processed: a crash before this line re-pulls the in-flight page
        # (safe, since the sink write is an idempotent upsert on `id`), and a
        # crash after this line resumes at `cursor`, never re-pulling a
        # completed page.
        checkpoints.append(cursor)

        if cursor is None:
            break

    return all_records, checkpoints, schema_alert

The retry loop treats a 429 as a distinct, expected condition, not a generic exception: it backs off exponentially with a cap and jitter, and gives up loudly after max_retries rather than looping forever. The schema check runs BEFORE the page is appended to results, so a schema change is caught on the very first affected page, and the function returns a structured alert naming exactly what changed instead of raising an opaque error or, worse, loading the mismatched rows. Retries now cover both RateLimited (429) and a new TransientError (5xx / connection failures): the original version only retried 429s, which silently contradicted the answer's own Trade-offs guidance below to 'retry a 429 or a 5xx'; a real 5xx would have crashed the extraction outright instead of backing off and retrying like the prose claimed.

Worked example

Running this against a fake API with 23 records across 3 pages, and a rate limiter that returns 429 on every 7th call, produces:

--- Run 1: full extraction, no schema change, verify checkpoint resume ---
records pulled: 23 (expected 23)
checkpoints written: [10, 20, None]
schema_alert: None

--- Run 2: resume from a mid-run checkpoint (simulating a crash after page 1) ---
records pulled after resume: 13 (expected 13)
resume correctness verified: first-10 + resumed = full set, no gaps or duplicates

--- Run 3: schema change mid-run is detected and halts rather than silently mis-loading ---
records pulled before halt: 0
schema_alert: {'cursor': None, 'expected': ['amount', 'id', 'updated_at'], 'observed': ['amount_cents', 'id', 'updated_at']}
schema-change halt verified: 0 rows silently mis-loaded; alert names exactly what changed

Run 2 restarts from the checkpoint saved after the first page (10) and pulls exactly the remaining 13 records, no more and no fewer, confirmed by checking that the first 10 records from run 1 plus the 13 resumed records equal the full 23-record set with no gaps or duplicates. Run 3 simulates the upstream API renaming amount to amount_cents: the very first page already has the new shape, so the function halts having loaded zero rows under the wrong schema, rather than silently writing an amount_cents value into a column your warehouse thinks is amount.

For the Redshift-sink variant of the same pattern, the upsert step at the end of each page becomes a staged COPY into a temp table followed by MERGE ... WHEN MATCHED THEN UPDATE ... WHEN NOT MATCHED THEN INSERT keyed on the record's own ID, and the same job would typically be scheduled and monitored by an orchestrator (Airflow or similar) that alerts on both a raised RuntimeError (retries exhausted) and a returned schema_alert (a human needs to look at the new field shape) as two distinct, separately-actionable failure conditions.

Trade-offs & pitfalls

• The checkpoint must be the cursor the API returns, not a locally-computed "now"; if you use "now" and a record was still being written when you polled, the next run's updated_since can silently skip it.
• Detecting a schema change by comparing keys on just the first record of a page is cheap but not exhaustive; a source that only changes the shape of some rows within a page needs a per-row check instead, at higher cost.
• Retrying "transient failures" broadly is a trap: retry a 429 or a 5xx, but do not blindly retry a 4xx that means "your query is malformed," since that will just fail identically forever and burn your retry budget on something backoff cannot fix.
• The idempotent-upsert assumption that makes "re-pull the in-flight page safely" true has to actually hold on the sink side; without a stable natural key on the destination table, a resumed pull after a crash can create duplicate rows instead of harmlessly overwriting them.

Direct answer

The core tension is that the backfill and the live sync are both consuming the same 10 requests/second budget from the same API, so the design has to explicitly partition that budget rather than let the two compete unpredictably, while making the backfill itself resumable and idempotent so a multi-day job surviving several restarts still converges on a correct, non-duplicated result.

Structured elaboration

Partitioning the rate-limit budget

• Reserve a fixed share of the 10 req/s for live sync (enough to keep it comfortably within its freshness service-level agreement (SLA)) and let the backfill use the remainder, rather than a naive "whoever gets there first" free-for-all between the two workloads.
• A shared, centralized rate limiter (not one limiter per worker) is required here: independent per-worker limiters cannot see each other's usage and will collectively exceed the source's real cap.

Parallelization strategy

• Parallelize across accounts, not within a single account's page sequence, since pages within one account's history typically must be walked in order for correct checkpointing, while different accounts are fully independent of each other.
• Size the worker pool to the rate-limit budget available to the backfill, not to raw compute capacity; more workers than the rate limit can support just means more of them sitting idle waiting for a token.

Checkpointing for resumability

• Checkpoint per account, independently: persist each account's furthest-completed page or cursor so a restart resumes only the accounts still in progress, not the ones already fully backfilled.
• Persist checkpoints frequently enough (after every page, not just at the end of an account) that a crash loses at most one page of already-fetched-but-uncommitted work per in-progress account.

Guaranteeing eventual consistency without duplicates

• Every record written by either the backfill or the live sync goes through the same idempotent upsert path, keyed on the record's own stable ID, so it does not matter which of the two processes writes a given record first or whether both happen to write it.
• Define a clear ordering rule for the rare case where both processes touch the same record concurrently (for example, "whichever write carries the later updated_at wins"), so the result is deterministic rather than a race.

Worked example

2,000 accounts need two years of history backfilled, at a shared 10 req/s budget, with 3 req/s reserved for live sync, leaving 7 req/s for the backfill. A pool of 7 backfill workers, each holding one token from a centralized rate limiter, is assigned accounts from a queue; each worker walks one account's full page history, checkpointing its cursor after every page, before picking up the next account from the queue. If the process crashes after 1,200 of the 2,000 accounts are fully done and a 1,201st is half-complete, a restart re-reads the checkpoint table, skips the 1,200 complete accounts entirely, resumes the 1,201st from its last saved cursor rather than page 1, and continues the remaining 799 from scratch. Throughout, both the backfill and the live sync write through the same upsert-by-record-ID path, so if live sync happens to ingest a very recent change to an account the backfill has not yet reached, the backfill's later arrival at that same record, carrying older data, does not overwrite the newer one, because the upsert conflict rule explicitly prefers the later updated_at.

Trade-offs & pitfalls

• Splitting the rate-limit budget statically (a fixed 3/7 split) is simple but can starve one workload if the actual demand shifts, for example if live sync traffic spikes; a smarter allocator that lets the backfill temporarily borrow idle live-sync capacity is more efficient but meaningfully more complex to build and reason about correctness for.
• Checkpointing at the account level but not the page level within an account means a crash can lose up to one full account's progress, not just one page; the finer-grained the checkpoint, the less repeated work a crash costs, at the price of more frequent writes to the checkpoint store.
• The "later updated_at wins" conflict rule assumes the source's own timestamps are trustworthy and consistent across accounts and over the two-year backfill window; if that assumption does not hold for some historical data, you need a different, explicit tie-breaking rule rather than silently trusting a timestamp that might be wrong.
• A single centralized rate limiter is also a single point of contention; at very high worker counts it can itself become a bottleneck, though at 7 concurrent workers against a 7 req/s budget this is not yet a real concern.

Direct answer

A consumer product at real scale typically ingests five recognizably different kinds of source: high-volume client behavioral events, infrastructure and CDN logs, lower-volume but business-critical transactional events (billing, membership), relatively small and slow-changing catalog or content metadata, and partner-reported measurement data whose format you do not control. Each has a distinct volume profile, cardinality, and downstream audience, and conflating them under one ingestion design is where a lot of real pipelines go wrong.

Structured elaboration

Client behavioral events

• Volume: the highest-volume source by a wide margin, often billions of events per day at real scale, generated continuously by every user session.
• Cardinality: very high on dimensions like user ID and session ID, moderate on event type (a bounded, known vocabulary of action names).
• Downstream consumers: personalization and recommendation systems needing near-real-time signal, plus analytics and experimentation platforms consuming it in batch.

CDN and infrastructure logs

• Volume: also very high, driven by request volume rather than user actions, and often noisier and less structured than application-level events.
• Cardinality: high on request-level dimensions (IP, URL, timestamp), but the record shape itself is usually simpler and more uniform than a rich behavioral event.
• Downstream consumers: operations and reliability teams for real-time monitoring, plus security teams for anomaly and abuse detection.

Billing and membership events

• Volume: orders of magnitude lower than behavioral events, since they correspond to discrete business transactions rather than continuous activity.
• Cardinality: lower on most dimensions, but each individual record carries much higher business stakes than a single behavioral event does.
• Downstream consumers: finance and revenue reporting, customer support (for account status lookups), and fraud detection.

Catalog or content metadata

• Volume: the lowest-volume and slowest-changing source of the group, updated on the order of the catalog's own size and change rate, not user activity.
• Cardinality: bounded by the size of the catalog itself, typically far smaller than any of the event-volume sources.
• Downstream consumers: the personalization and search systems that join it against behavioral events, plus the product surfaces that render it directly.

Partner-reported measurement

• Volume: modest and typically batch-delivered on the partner's own schedule (daily or weekly files), not continuously streamed.
• Cardinality: depends heavily on the specific partner and measurement type, but the defining trait is that its format and delivery schedule are entirely outside your control.
• Downstream consumers: business reporting, revenue reconciliation, and any feature that specifically depends on that partner's data.

Worked example

A large streaming platform's product, personalization, and operations functions draw on exactly this mix: playback and interaction events from every viewing session (the highest-volume behavioral source, feeding both real-time personalization and batch analytics), CDN delivery logs (feeding operational dashboards and anomaly detection for stream quality issues), billing and subscription events (lower volume, high business stakes, feeding revenue reporting), catalog and content metadata (title, genre, cast, availability windows, feeding search and recommendation), and partner-reported measurement from advertising or co-production partners (batch-delivered, feeding revenue-share reconciliation). A team designing ingestion for this platform that treated all five as "just events to ingest" with one uniform pipeline would badly under-serve the billing source's correctness requirements while badly over-engineering the catalog source's freshness needs, since the two have almost nothing in common except both technically being "data."

Trade-offs & pitfalls

• The biggest real mistake is applying one uniform service-level agreement (SLA) and one uniform pipeline design across all five source types; a design tuned for the highest-volume behavioral stream is usually the wrong shape for the lowest-volume, highest-stakes billing stream, and vice versa.
• Cardinality is easy to underestimate for behavioral data specifically; a naive schema or index design that works fine at prototype scale can fail badly once real user-ID and session-ID cardinality is at production volume.
• Partner-reported measurement is the source most likely to have unannounced format drift, precisely because you have the least influence over the partner's own release process; it deserves proportionally more ingestion-time validation than its modest volume alone would suggest.
• Do not assume "downstream consumer" is singular for any of these; behavioral events in particular routinely feed both a real-time system (personalization) and a batch system (analytics) with genuinely different freshness needs from the same underlying stream.

Direct answer

A token bucket is a simple rate-limiting mechanism: it holds up to some fixed number of tokens, refilled continuously at the source's allowed rate, each call must take one token before it is allowed to fire, and if the bucket is empty the caller waits for the next trickle of refill rather than proceeding, which is what naturally caps you at the source's rate. A token-bucket limiter refilled at exactly the source's rate (100 tokens/minute) will genuinely respect the cap, but only if it starts EMPTY rather than full; a bucket that starts full lets the worker burn its whole capacity instantly and then refill more on top, so a 60-second window straddling that initial burst can see up to double the real cap. Below is a runnable worker, with its own default set to the SAFE (empty-start) behavior for this specific hard-cap use case, verified by executing both the unsafe and the safe configuration end to end.

Structured elaboration

from fractions import Fraction as F


class VirtualClock:
    def __init__(self):
        self.t = F(0)

    def now(self):
        return self.t

    def advance(self, seconds):
        self.t += seconds


class TokenBucketRateLimiter:
    # start_full defaults to False here deliberately: this class exists specifically to
    # respect a HARD external cap, and defaulting to a full initial bucket is exactly the
    # bug this answer is about (see the Worked example). A caller who genuinely wants the
    # textbook burst-tolerant behavior must opt in explicitly with start_full=True.
    def __init__(self, clock, capacity=100, refill_per_second=F(100, 60), start_full=False):
        self.clock = clock
        self.capacity = F(capacity)
        self.tokens = self.capacity if start_full else F(0)
        self.refill_rate = F(refill_per_second)
        self.last_refill = clock.now()

    def _refill(self):
        elapsed = self.clock.now() - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = self.clock.now()

    def try_acquire(self):
        self._refill()
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

    def seconds_until_next_token(self):
        self._refill()
        if self.tokens >= 1:
            return F(0)
        return (1 - self.tokens) / self.refill_rate


def sync_worker(records, limiter, clock, batch_size=10, call_log=None):
    call_log = call_log if call_log is not None else []
    i = 0
    n = len(records)
    while i < n:
        if limiter.try_acquire():
            batch = records[i:i + batch_size]
            call_log.append(clock.now())
            i += len(batch)
        else:
            clock.advance(limiter.seconds_until_next_token())
    return call_log

The 50,000 records themselves form the work queue: in this single-process example it is simply the records list, drained batch by batch by sync_worker, but the same design generalizes directly to a distributed work queue (SQS, Redis, or a database-backed job table) feeding a pool of worker processes, with the token-bucket state moved into a shared store (Redis again, typically) so every worker draws from the same rate budget instead of each worker getting its own 100-per-minute allowance and collectively blowing through the source's real cap. Batching 10 records per call means syncing 50,000 records costs exactly 5,000 API calls. The worker only ever calls the source when try_acquire succeeds; otherwise it advances straight to the moment the next token becomes available (a real worker would sleep that duration) instead of busy-polling.

Worked example

def worst_window(call_log, window=60):
    call_log = sorted(call_log)
    worst = 0
    j = 0
    for i in range(len(call_log)):
        while call_log[i] - call_log[j] >= window:
            j += 1
        worst = max(worst, i - j + 1)
    return worst


records = list(range(50000))
batch_size = 10

print("--- Version A: bucket starts FULL (start_full=True, the textbook default, opted into explicitly) ---")
clock_a = VirtualClock()
limiter_a = TokenBucketRateLimiter(clock_a, start_full=True)
log_a = sync_worker(records, limiter_a, clock_a, batch_size=batch_size)
wa = worst_window(log_a)
print(f"total API calls: {len(log_a)}, worst 60s window: {wa} (cap is 100)")

print("\n--- Version B: bucket starts EMPTY (start_full=False, the class's own default) ---")
clock_b = VirtualClock()
limiter_b = TokenBucketRateLimiter(clock_b)  # no override needed: False is now the default
log_b = sync_worker(records, limiter_b, clock_b, batch_size=batch_size)
wb = worst_window(log_b)
elapsed = clock_b.now()
print(f"total records synced: {len(records)}")
print(f"total API calls made: {len(log_b)} (50,000 / batch_size=10 = {len(records)//batch_size} expected)")
print(f"elapsed virtual time: {float(elapsed)}s ({float(elapsed)/60:.4f} minutes)")
print(f"worst calls observed in any 60s sliding window: {wb} (cap is 100)")
print(f"exact fraction elapsed equals 3000: {elapsed == 3000}")

Actual output from running the above:

--- Version A: bucket starts FULL (start_full=True, the textbook default, opted into explicitly) ---
total API calls: 5000, worst 60s window: 199 (cap is 100)

--- Version B: bucket starts EMPTY (start_full=False, the class's own default) ---
total records synced: 50000
total API calls made: 5000 (50,000 / batch_size=10 = 5000 expected)
elapsed virtual time: 3000.0s (50.0000 minutes)
worst calls observed in any 60s sliding window: 100 (cap is 100)
exact fraction elapsed equals 3000: True

Version A exceeds the cap: a full initial bucket lets a 60-second window straddling t=0 see up to 199 calls against a cap of 100, a genuine two-times overshoot, not a rounding artifact. Version B, using the class's own default (start_full=False), lands on exactly 100 calls in the worst window and exactly 3000.0 seconds total (Fraction(3000, 1), not merely close to 3000.0 as a float), both matching the theoretical prediction (50,000 records / 10 per batch = 5,000 calls; 5,000 calls at 100/minute = 50 minutes = 3,000 seconds).

This run surfaced three real bugs in sequence during development, each caught only by execution, and each fixed in the version shown above rather than only described. First, comparing accumulated float tokens against >= 1 never becomes true (0.6 * (100/60) == 0.9999999999999999 in IEEE-754 float, not exactly 1.0), so an early float-based version of this worker spun without terminating; switching all token/time accounting to fractions.Fraction (as shown) removes that failure mode entirely rather than working around it with a tolerance band. Second, before the start_full default was set to False, the class's OWN default silently reproduced the two-times overshoot shown as Version A above: a caller who simply wrote TokenBucketRateLimiter(clock) with no other arguments got the unsafe behavior with no warning. Third, an intermediate rational-but-not-yet-final version of the clock still accumulated small drift across 5,000 sequential refills, landing on 101 calls in the worst window instead of 100; using exact Fraction arithmetic throughout (not just for the rate, but for every advance() call) removed that drift entirely, and the final run above lands on precisely 100 and precisely 3000.0 with zero drift.

Trade-offs & pitfalls

• Starting a token bucket full is the standard, textbook implementation, and it is wrong as this class's DEFAULT specifically because the requirement is to never exceed the SOURCE's cap in any window, not to maximize your own burst; a caller who genuinely wants the textbook burst-tolerant behavior for some other use case can still opt in with start_full=True, but the safe behavior should never be something a caller has to remember to ask for.
• Floating-point time accumulation drift is invisible at small scale and only surfaces after thousands of iterations; a code review reading this logic would very plausibly not catch it, which is exactly why it needed to be executed at the real scale (5,000 calls) rather than sanity-checked on a handful.
• A rate limiter that "mostly" respects a hard external cap is not a safe design; a source enforcing a strict per-minute cap can and does suspend or ban an integration that occasionally exceeds it, so "off by one call in a window" is a real production risk, not a rounding curiosity.
• Using exact rational arithmetic for time accounting is more code than floats and is unnecessary for most application logic, but it is the right tool specifically where a long-running process accumulates many small time increments against a hard external constraint; know when the extra rigor is worth it.