Beginner

Data Extraction for Machine Learning

Learn how to extract data from diverse sources - databases, APIs, files, and real-time streams - to feed your ML pipelines.

Extraction Sources Overview

ML projects typically pull data from multiple sources. Understanding the characteristics of each source helps you design reliable extraction pipelines:

Source Type	Examples	Extraction Pattern	ML Use Case
Relational DBs	PostgreSQL, MySQL	SQL queries, CDC	Structured features
NoSQL DBs	MongoDB, DynamoDB	Collection scans, change streams	Unstructured data
REST APIs	Third-party services	Paginated requests	External signals
File Systems	S3, GCS, HDFS	Batch reads, glob patterns	Logs, images, text
Event Streams	Kafka, Kinesis	Consumer groups	Real-time features
Web Scraping	HTML pages	Parsers, headless browsers	Training data collection

Database Extraction

Databases are the most common data source for ML. There are two primary extraction patterns:

Full Extraction

Pull the entire table on every run. Simple but expensive for large tables. Best for small reference tables or initial loads.

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine("postgresql://user:pass@host:5432/db")

# Full extraction
df = pd.read_sql("SELECT * FROM customers", engine)
print(f"Extracted {len(df)} rows")

Incremental Extraction

Extract only new or changed records since the last run. Uses timestamps or change data capture (CDC) for efficiency.

# Incremental extraction using watermark
last_watermark = get_last_watermark()  # stored from previous run

query = """
    SELECT * FROM transactions
    WHERE updated_at > %(watermark)s
    ORDER BY updated_at
"""
new_records = pd.read_sql(query, engine, params={"watermark": last_watermark})
save_watermark(new_records["updated_at"].max())

API Extraction

When extracting from APIs, handle pagination, rate limits, retries, and authentication:

import requests
import time

def extract_from_api(base_url, api_key, max_pages=100):
    all_records = []
    page = 1

    while page <= max_pages:
        response = requests.get(
            f"{base_url}?page={page}&per_page=100",
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30
        )

        if response.status_code == 429:  # Rate limited
            time.sleep(int(response.headers.get("Retry-After", 60)))
            continue

        response.raise_for_status()
        data = response.json()

        if not data["results"]:
            break

        all_records.extend(data["results"])
        page += 1

    return all_records

File-Based Extraction

Cloud storage is a common landing zone for ML data. Use partitioned file layouts for efficient extraction:

import pyarrow.parquet as pq
import s3fs

# Read partitioned Parquet files from S3
fs = s3fs.S3FileSystem()
dataset = pq.ParquetDataset(
    "s3://ml-data/features/date=2025-01-15/",
    filesystem=fs
)
df = dataset.read().to_pandas()

Stream Extraction

For real-time ML features, extract data from event streams like Apache Kafka:

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "user-events",
    bootstrap_servers=["kafka:9092"],
    group_id="ml-feature-pipeline",
    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
    auto_offset_reset="latest"
)

for message in consumer:
    event = message.value
    process_event_for_features(event)

Extraction Best Practices

• Idempotency: Design extractions that can be safely re-run without duplicating data.
• Checkpointing: Store extraction progress (watermarks, offsets) so pipelines can resume after failures.
• Parallelism: Split large extractions into chunks and process them concurrently.
• Source isolation: Use read replicas for database extraction to avoid impacting production systems.
• Logging: Record extraction metadata (row counts, timestamps, source versions) for debugging and auditing.