3. Data acquisition and ingestion

Every applied machine learning system begins with data that must be acquired, transported, and reshaped into a form a model can consume. This chapter treats ingestion as an engineering discipline rather than a one-off script. The goal is to move raw records from heterogeneous sources into a reproducible, typed, and versioned dataset while preserving enough provenance to debug failures later.

We can group sources into five families: APIs (structured HTTP endpoints), web scraping (unstructured HTML), databases (SQL/NoSQL), file formats (CSV, JSON, Parquet), and streaming sources (Kafka, Kinesis, WebSockets). Each family has distinct failure modes, cost profiles, and schema guarantees. A robust ingestion layer abstracts these differences behind a common contract: read a batch or event, validate it, and write it to a landing zone.

APIs: Pagination, Rate Limits, and Idempotency

REST APIs are the most common structured source. The critical engineering concerns are pagination, rate limiting, retries, and idempotency. Consider a hypothetical public dataset API that returns records in pages of 100 with a cursor token. A naive loop that ignores the cursor will silently truncate the dataset. A production-grade fetch should persist the cursor so a failed run can resume without re-downloading everything.

import requests, time, json

BASE = "https://api.example-dataset.org/v2/records"
cursor = None
records = []

while True:
    params = {"limit": 100}
    if cursor:
        params["cursor"] = cursor
    resp = requests.get(BASE, params=params, timeout=30)
    if resp.status_code == 429:
        time.sleep(int(resp.headers.get("Retry-After", 5)))
        continue
    resp.raise_for_status()
    payload = resp.json()
    records.extend(payload["data"])
    cursor = payload.get("next_cursor")
    if not cursor:
        break

with open("raw/records.jsonl", "w") as f:
    for r in records:
        f.write(json.dumps(r) + "\n")

Two details matter here. First, Retry-After is honored rather than guessed, which avoids escalating a soft rate limit into a hard ban. Second, writing JSONL (one JSON object per line) instead of a single JSON array makes the output append-friendly and streamable. If the job crashes mid-write, the partial file is still parseable line by line.

Web Scraping: Structure Without a Contract

Scraping is appropriate when no API exists and the terms of service permit automated access. The main risk is brittleness: a CSS selector that works today may break after a site redesign. Prefer semantic anchors (stable IDs, data attributes) over positional selectors like div:nth-child(3). Always respect robots.txt, set a descriptive User-Agent, and throttle requests.

from bs4 import BeautifulSoup
import requests, time

resp = requests.get("https://example-shop.org/catalog",
                    headers={"User-Agent": "ml-course-bot/1.0"}, timeout=20)
soup = BeautifulSoup(resp.text, "html.parser")

rows = []
for card in soup.select("article.product[data-sku]"):
    rows.append({
        "sku": card["data-sku"],
        "title": card.select_one("h2.title").get_text(strip=True),
        "price": float(card.select_one("span.price")["data-value"]),
    })
    time.sleep(0.5)

Note the use of data-sku and data-value: these attributes are typically stable because they drive the site's own JavaScript. When scraping at scale, store the raw HTML alongside the parsed fields so you can re-parse without re-fetching if your extraction logic changes.

Databases: Pull Versus Push

Relational databases remain the backbone of enterprise ML. Two ingestion patterns dominate. Batch pull uses a query with a watermark column (e.g., updated_at) to fetch only new or changed rows. Change data capture (CDC) streams the database's write-ahead log so downstream systems see every insert, update, and delete in order. Batch pull is simpler and sufficient for daily retraining; CDC is necessary when latency matters or when deletes must be propagated.

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("postgresql://user:pass@db:5432/app")
query = """
    SELECT order_id, customer_id, amount, updated_at
    FROM orders
    WHERE updated_at > %(watermark)s
    ORDER BY updated_at
"""
df = pd.read_sql(query, engine, params={"watermark": "2024-05-01T00:00:00Z"})
df.to_parquet("landing/orders.parquet", index=False)

The watermark must be advanced only after the write succeeds. If you update the watermark before the Parquet file is durable, a crash loses data permanently. This is the classic at-least-once versus exactly-once trade-off: at-least-once with idempotent writes is usually the pragmatic choice.

File Formats: CSV, JSON, and Parquet

CSV is universal but untyped, has no schema, and encodes everything as text. JSON handles nesting and is ideal for API payloads, but is verbose and slow to parse at scale. Parquet is a columnar, compressed, typed format with embedded schema metadata. For analytical workloads, Parquet typically reduces storage by 5–10x and speeds up column-selective reads by an order of magnitude compared to CSV.

A practical rule: land raw data as JSONL or CSV for auditability, then convert to Parquet for training. Keep both. The raw layer is your replay log; the Parquet layer is your fast path.

Streaming Sources: Windows and Watermarks

Streaming ingestion (Kafka, Kinesis, WebSockets) delivers events continuously. The central challenge is not reading the stream but deciding when a window of events is complete. Event time differs from processing time due to network delays and out-of-order arrival. A watermark is the system's assertion that no events older than a given timestamp will arrive. Late events beyond the watermark are either dropped or routed to a side output.

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "orders",
    bootstrap_servers="broker:9092",
    group_id="ml-ingest",
    auto_offset_reset="earliest",
    enable_auto_commit=False,
    value_deserializer=lambda b: json.loads(b.decode("utf-8")),
)

for msg in consumer:
    event = msg.value
    write_to_landing(event)
    consumer.commit()  # commit only after durable write

Disabling auto-commit and committing manually after the write gives at-least-once delivery. Combined with a deduplication key (e.g., event_id) in the landing zone, this yields effectively-once semantics for downstream consumers.

Putting It Together: A Minimal Ingestion Contract

Regardless of source, every ingestion job should satisfy four properties. Reproducibility: given the same inputs and watermark, it produces the same output. Idempotency: re-running a batch does not duplicate records. Observability: row counts, null rates, and schema hashes are logged per run. Recoverability: a crash mid-run can resume without data loss or duplication.

With sources connected and a landing zone in place, the next stage is validation and transformation. The ingestion layer's job is to deliver trustworthy raw material; it should not attempt feature engineering or model-specific reshaping. Keeping that boundary clean makes both layers easier to test and evolve independently.