Data Ingestion Patterns: Batch, Streaming, CDC, and APIs

Data Ingestion Patterns: Bringing Data into the Pipeline

Data ingestion is the process of moving data from source systems into a data platform for processing, analysis, and storage.

The Ingestion Landscape

Data Sources:

• Relational databases — PostgreSQL, MySQL
• NoSQL stores — MongoDB, Cassandra
• SaaS APIs — Salesforce, HubSpot
• Streaming platforms — Kafka, Kinesis
• File systems — SFTP, S3
• Message queues — RabbitMQ, SQS

Key Considerations:

1. Source capabilities — what does the source support?
2. Data volume — how much data needs to be ingested?
3. Latency requirements — real-time or batch?
4. Impact tolerance — can the source handle the load?

Key Insight: Each source demands a different ingestion strategy based on its capabilities, data volume, latency requirements, and impact tolerance.

Data Ingestion Patterns Overview

CDC Flow Diagram

Architecture Diagram

Key Concepts

Production Code

CDC Pipeline with Debezium and Kafka

from confluent_kafka import Consumer, Producer
from confluent_kafka.admin import AdminClient, NewTopic
import json
import logging
from typing import Dict, List
from dataclasses import dataclass, asdict
from datetime import datetime

logger = logging.getLogger(__name__)


@dataclass
class CDCEvent:
    """Represents a Change Data Capture event from Debezium."""
    operation: str  # c=create, u=update, d=delete
    before: Dict
    after: Dict
    source: Dict
    ts_ms: int
    table: str


class DebeziumCDCProcessor:
    """Process CDC events from Debezium via Kafka."""

    def __init__(self, kafka_config: Dict, topic: str):
        self.topic = topic
        self.consumer = Consumer({
            "bootstrap.servers": kafka_config["bootstrap.servers"],
            "group.id": "cdc-processor",
            "auto.offset.reset": "earliest",
            "enable.auto.commit": False,
        })
        self.producer = Producer({
            "bootstrap.servers": kafka_config["bootstrap.servers"],
            "acks": "all",
            "enable.idempotence": True,
        })

    def parse_cdc_event(self, raw_value: str) -> CDCEvent:
        """Parse Debezium CDC envelope format."""
        envelope = json.loads(raw_value)
        return CDCEvent(
            operation=envelope["op"],
            before=envelope.get("before"),
            after=envelope.get("after"),
            source=envelope["source"],
            ts_ms=envelope["ts_ms"],
            table=envelope["source"]["table"],
        )

    def process_create(self, event: CDCEvent) -> Dict:
        """Handle new record creation."""
        return {
            "action": "insert",
            "table": event.table,
            "data": event.after,
            "cdc_timestamp": datetime.fromtimestamp(event.ts_ms / 1000),
        }

    def process_update(self, event: CDCEvent) -> Dict:
        """Handle record update with change detection."""
        changes = {}
        if event.before and event.after:
            for key in event.after:
                if event.before.get(key) != event.after.get(key):
                    changes[key] = {
                        "old": event.before.get(key),
                        "new": event.after.get(key),
                    }
        return {
            "action": "update",
            "table": event.table,
            "data": event.after,
            "changes": changes,
            "cdc_timestamp": datetime.fromtimestamp(event.ts_ms / 1000),
        }

    def process_delete(self, event: CDCEvent) -> Dict:
        """Handle record deletion."""
        return {
            "action": "delete",
            "table": event.table,
            "data": event.before,
            "cdc_timestamp": datetime.fromtimestamp(event.ts_ms / 1000),
        }

    def run(self, max_events: int = None):
        """Main consumption loop for CDC events."""
        self.consumer.subscribe([self.topic])
        event_count = 0

        try:
            while True:
                msg = self.consumer.poll(timeout=1.0)
                if msg is None:
                    continue
                if msg.error():
                    logger.error(f"Consumer error: {msg.error()}")
                    continue

                event = self.parse_cdc_event(msg.value().decode("utf-8"))

                if event.operation == "c":
                    result = self.process_create(event)
                elif event.operation == "u":
                    result = self.process_update(event)
                elif event.operation == "d":
                    result = self.process_delete(event)
                else:
                    logger.warning(f"Unknown operation: {event.operation}")
                    continue

                # Produce processed event to downstream topic
                self.producer.produce(
                    topic=f"{self.topic}.processed",
                    key=event.after.get("id", "").encode() if event.after else None,
                    value=json.dumps(result, default=str).encode(),
                )

                event_count += 1
                if max_events and event_count >= max_events:
                    break

                if event_count % 1000 == 0:
                    logger.info(f"Processed {event_count} CDC events")
                    self.consumer.commit()

        finally:
            self.consumer.commit()
            self.consumer.close()
            self.producer.flush()


# Debezium connector configuration (JSON)
DEBEZIUM_CONFIG = {
    "name": "postgres-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname": "postgres-host",
        "database.port": "5432",
        "database.user": "debezium",
        "database.password": "${secrets:debezium-password}",
        "database.dbname": "production",
        "database.server.name": "production",
        "plugin.name": "pgoutput",
        "slot.name": "debezium_slot",
        "publication.name": "debezium_publication",
        "table.include.list": "public.orders,public.customers",
        "tombstones.on.delete": "true",
        "transforms": "route",
        "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
        "transforms.route.replacement": "$3",
    },
}

Batch Ingestion with Deduplication

import pandas as pd
import hashlib
from datetime import datetime, timedelta
from typing import Optional
import logging

logger = logging.getLogger(__name__)


class BatchIngestor:
    """Batch ingestion with idempotent deduplication and upsert operations."""

    def __init__(self, target_connection, dedup_key: str = "id"):
        self.target_connection = target_connection
        self.dedup_key = dedup_key

    def extract_with_watermark(
        self,
        source_connection,
        query: str,
        watermark_table: str,
        last_watermark: Optional[datetime] = None,
    ) -> pd.DataFrame:
        """Extract data from source using watermark for incremental loading."""
        if last_watermark is None:
            # First run: get the max value from the target
            watermark_query = f"SELECT MAX(updated_at) FROM {watermark_table}"
            last_watermark = pd.read_sql(watermark_query, self.target_connection).iloc[0, 0]

        if last_watermark is None:
            # No existing data: full load
            return pd.read_sql(query, source_connection)

        # Incremental: only rows modified since last watermark
        incremental_query = f"""
            {query}
            WHERE updated_at > '{last_watermark.isoformat()}'
        """
        logger.info(f"Extracting incremental data since {last_watermark}")
        return pd.read_sql(incremental_query, source_connection)

    def compute_record_hash(self, df: pd.DataFrame) -> pd.DataFrame:
        """Compute a hash of all columns for change detection."""
        hash_columns = [col for col in df.columns if col != self.dedup_key]
        df["_record_hash"] = df[hash_columns].apply(
            lambda row: hashlib.md5(
                "|".join(str(v) for v in row).encode()
            ).hexdigest(),
            axis=1,
        )
        return df

    def upsert(self, df: pd.DataFrame, target_table: str) -> dict:
        """Perform idempotent upsert (insert-or-update) operation."""
        if df.empty:
            return {"inserted": 0, "updated": 0, "unchanged": 0}

        df = self.compute_record_hash(df)

        # Read existing records for comparison
        existing_query = f"""
            SELECT {self.dedup_key}, _record_hash
            FROM {target_table}
            WHERE {self.dedup_key} IN ({','.join(df[self.dedup_key].astype(str))})
        """
        try:
            existing = pd.read_sql(existing_query, self.target_connection)
        except Exception:
            existing = pd.DataFrame(columns=[self.dedup_key, "_record_hash"])

        # Merge existing hashes with new data
        merged = df.merge(
            existing, on=self.dedup_key, how="left", suffixes=("", "_existing")
        )

        # Classify records
        new_records = merged[merged["_record_hash_existing"].isna()]
        changed_records = merged[
            (merged["_record_hash_existing"].notna()) &
            (merged["_record_hash"] != merged["_record_hash_existing"])
        ]
        unchanged_records = merged[
            (merged["_record_hash_existing"].notna()) &
            (merged["_record_hash"] == merged["_record_hash_existing"])
        ]

        # Insert new records
        if not new_records.empty:
            insert_df = new_records.drop(columns=["_record_hash_existing"])
            insert_df.to_sql(
                target_table,
                self.target_connection,
                if_exists="append",
                index=False,
                method="multi",
                chunksize=500,
            )
            logger.info(f"Inserted {len(new_records)} new records")

        # Update changed records (using temporary table for efficiency)
        if not changed_records.empty:
            update_df = changed_records.drop(columns=["_record_hash_existing"])
            temp_table = f"{target_table}_temp_update"
            update_df.to_sql(
                temp_table,
                self.target_connection,
                if_exists="replace",
                index=False,
            )
            # Execute MERGE/UPSERT via SQL
            merge_query = f"""
                INSERT INTO {target_table}
                SELECT * FROM {temp_table}
                ON CONFLICT ({self.dedup_key})
                DO UPDATE SET
                    {', '.join(
                        f"{col} = EXCLUDED.{col}"
                        for col in update_df.columns
                        if col != self.dedup_key
                    )}
            """
            self.target_connection.execute(merge_query)
            self.target_connection.execute(f"DROP TABLE {temp_table}")
            logger.info(f"Updated {len(changed_records)} changed records")

        return {
            "inserted": len(new_records),
            "updated": len(changed_records),
            "unchanged": len(unchanged_records),
        }

    def update_watermark(self, watermark_table: str, max_timestamp: datetime):
        """Update the watermark to track ingestion progress."""
        watermark_df = pd.DataFrame([{
            "pipeline": watermark_table,
            "last_watermark": max_timestamp,
            "updated_at": datetime.utcnow(),
        }])
        watermark_df.to_sql(
            "ingestion_watermarks",
            self.target_connection,
            if_exists="append",
            index=False,
        )
        logger.info(f"Watermark updated to {max_timestamp}")


# Usage
ingestor = BatchIngestor(target_connection=engine, dedup_key="order_id")
df = ingestor.extract_with_watermark(
    source_connection=source_engine,
    query="SELECT * FROM orders",
    watermark_table="orders_fact",
)
results = ingestor.upsert(df, target_table="orders_fact")
logger.info(f"Ingestion results: {results}")

Best Practices

1. Implement idempotent ingestion — use deduplication keys and upsert/MERGE operations so re-runs produce identical results.
2. Use CDC (Debezium) for relational databases when real-time freshness and delete capture are required.
3. Set watermarks to track ingestion progress and detect gaps. Alert when gaps exceed SLA thresholds.
4. Use schema registry at ingestion time to reject malformed data before it enters the pipeline.
5. Implement dead letter queues for records that fail ingestion. Never silently drop failed records.
6. Rate-limit API polling to respect source system rate limits. Use exponential backoff for 429 responses.
7. Verify webhook signatures to prevent unauthorized data injection. Use HMAC-SHA256 for signature validation.
8. Monitor source system impact — CDC should have negligible impact; polling should not exceed 10% of source CPU.
9. Design for backfill — support full reprocessing from raw data without modifying the ingestion pipeline.
10. Document data contracts with source system owners: expected schemas, update frequencies, and SLAs.

Ingestion Pattern Selection Matrix

CDC Tool Comparison

See Also

• 019 - Apache Kafka: Topics, Producers, and Consumers - Kafka as ingestion backbone
• 016 - ETL vs ELT - Transformation paradigm after ingestion
• 025 - Data Quality: Validation Frameworks - Quality checks at ingestion
• 022 - Spark Structured Streaming - Streaming ingestion processing
• 028 - Error Handling, Retries, and Dead Letter Queues - Error handling for ingestion