3. Data source integration

An analytics platform is only as trustworthy as the data it can reliably acquire. Data source integration is the discipline of moving data from operational systems into the platform's landing and processing zones while preserving correctness, ordering, and timeliness. This chapter examines the four dominant ingestion families — batch connectors, streaming connectors, change data capture (CDC), and API/file-based ingestion — and explains how to choose among them, how each behaves under failure, and what parameters actually matter in production.

Batch vs. Streaming: Choosing the Right Transport

Batch ingestion reads a bounded set of records on a schedule (hourly, nightly) and writes them as a discrete load. Streaming ingestion consumes an unbounded, continuously arriving event stream and processes records as they appear. The decision is not about "modern vs. legacy" — it is about the freshness requirement and the cost of reprocessing.

A nightly batch job that loads 40 million order rows into a warehouse is perfectly appropriate when downstream dashboards refresh daily. A fraud-scoring pipeline that must react within two seconds cannot wait for a batch window; it needs a streaming connector with sub-second end-to-end latency. The practical rule: choose the slowest transport that still satisfies the business freshness SLA, because batch is cheaper to operate, easier to replay, and simpler to reason about.

Change Data Capture (CDC)

CDC extracts row-level inserts, updates, and deletes from a source database's transaction log rather than polling the table. For PostgreSQL this means reading the logical replication slot via the pgoutput plugin; for MySQL it means consuming the binary log (binlog) with ROW format enabled. The advantage over timestamp-based polling is that CDC captures deletes and intermediate updates that a WHERE updated_at > :last_run query would silently miss.

A minimal Debezium-style PostgreSQL connector configuration looks like this:

{
  "name": "orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg-primary.internal",
    "database.port": "5432",
    "database.dbname": "commerce",
    "database.server.name": "commerce",
    "table.include.list": "public.orders,public.order_items",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_orders",
    "publication.autocreate.mode": "filtered",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "true",
    "heartbeat.interval.ms": "10000"
  }
}

Two parameters deserve attention. snapshot.mode=initial performs a consistent snapshot of existing rows before streaming changes, so the target starts with full history. heartbeat.interval.ms emits periodic heartbeat records on an otherwise idle slot; without it, PostgreSQL retains WAL segments indefinitely and the primary disk fills up — a classic CDC failure mode. The replication slot must also be monitored: if the consumer falls behind, pg_replication_slots.active goes false and restart_lsn stops advancing.

API Ingestion

Many SaaS systems expose data only through REST or GraphQL APIs. API ingestion is pull-based and must handle three realities: pagination, rate limits, and incremental cursors. A well-behaved connector stores the last successful cursor (a timestamp, an opaque page token, or an ETag) and resumes from it on the next run.

Consider a Stripe-style charges endpoint. The connector issues a request with a cursor parameter, follows has_more until exhausted, and persists the final object ID:

GET /v1/charges?limit=100&starting_after=ch_3PxK...&created[gte]=1717200000
Authorization: Bearer sk_live_***
Idempotency-Key: charges-20240601-001

# Response
{
  "object": "list",
  "data": [ { "id": "ch_3PxK...", "amount": 4200, ... } ],
  "has_more": true
}

Rate limiting is handled with exponential backoff on HTTP 429, respecting the Retry-After header. Because APIs can return duplicate records after a retry, the landing table should carry a natural key (the provider's object ID) and use an upsert or MERGE to guarantee idempotency. Without this, a single retried page can double-count revenue.

File-Based Ingestion Patterns

File ingestion covers SFTP drops, object-storage prefixes, and partner-delivered CSV/Parquet exports. The two dominant patterns are full-snapshot replacement and incremental append. Full-snapshot loads are simple but expensive at scale; incremental appends require a reliable partition key (usually a date) and a manifest to avoid re-reading already-processed files.

A robust object-storage pattern uses a manifest table to track processed files and their checksums:

-- landing.ingest_manifest
file_path            TEXT PRIMARY KEY,
etag                 TEXT,
row_count            BIGINT,
ingested_at          TIMESTAMPTZ,
status               TEXT  -- 'pending' | 'loaded' | 'failed'

-- Idempotent load guard
INSERT INTO landing.ingest_manifest (file_path, etag, status)
VALUES ('s3://raw/orders/dt=2024-06-01/part-000.parquet', 'a1b2c3', 'pending')
ON CONFLICT (file_path) DO NOTHING;

Only files whose etag differs from the stored value are re-read. This makes the pipeline safe to re-run after a partial failure: already-loaded files are skipped, and the failed file is retried. For late-arriving data, partition by event date rather than arrival date so backfills land in the correct partition.

Unifying the Four Families

Regardless of transport, every ingestion path should converge on the same landing contract: an immutable raw layer, a persisted offset or cursor, and an idempotency key. This uniformity is what allows a single replay mechanism to serve batch backfills, CDC re-snapshots, API re-pulls, and file re-reads. When a downstream model is found to be wrong, the fix is to reset the relevant offset and re-run — not to manually patch the warehouse.

The next design decision — how to model and transform this landed data — depends on whether the raw layer preserves enough fidelity to reconstruct any downstream state. That is the subject of the following chapter on transformation and modeling layers.