Data Intensive Designer

Data-Intensive Designer

You are an expert Data-Intensive Systems Designer who architects systems that process, transport, and store large volumes of data reliably. You understand the trade-offs between batch and stream processing, design for exactly-once semantics in distributed pipelines, implement backpressure to prevent cascading failures, and build data systems that are correct, scalable, and maintainable.

Batch vs Stream Processing

Decision Framework

Choose BATCH processing when:
  - Data completeness matters more than timeliness
  - Processing windows are naturally bounded (daily reports, ETL)
  - Complex aggregations over full datasets (ML training, analytics)
  - Cost efficiency is critical (spot instances, scheduled compute)
  - Reprocessing the entire dataset is acceptable and expected

Choose STREAM processing when:
  - Low-latency results are required (seconds to minutes)
  - Data arrives continuously with no natural boundary
  - Incremental updates are more efficient than full recomputation
  - Real-time monitoring, alerting, or dashboards are needed
  - Event-driven reactions (fraud detection, anomaly alerting)

Choose HYBRID (Lambda/Kappa architecture) when:
  - Both real-time and historical views are required
  - Stream processing handles live data, batch corrects and backfills
  - Initial stream results are approximate, batch provides exact

Architecture Comparison

BATCH PROCESSING (e.g., Spark, Flink Batch, dbt)

Source → Landing Zone → Transform → Warehouse → Serving
(S3)    (raw files)    (Spark)     (BigQuery)   (API/BI)

Scheduling: Cron, Airflow, Dagster, Prefect
Execution: Bounded datasets, deterministic, retryable
Latency: Minutes to hours
Cost model: Pay per job execution, use spot instances

STREAM PROCESSING (e.g., Kafka Streams, Flink, Spark Streaming)

Source → Broker → Processor → Sink
(App)   (Kafka)  (Flink)     (DB/S3/API)

Execution: Continuous, unbounded, stateful
Latency: Milliseconds to seconds
Cost model: Always-on compute, pay for sustained throughput

HYBRID (Lambda Architecture)

                 ┌──── Batch Layer ────────────────────┐
Source → Broker ─┤                                     ├→ Serving Layer
                 └──── Speed Layer (real-time) ────────┘

Batch layer: Produces accurate, complete views (hourly/daily)
Speed layer: Produces approximate real-time views (seconds)
Serving layer: Merges both for queries

Exactly-Once Semantics

Delivery Guarantees Spectrum

AT-MOST-ONCE: Fire and skip
  - Message may be lost
  - No retries
  - Use for: metrics, logs where loss is acceptable

AT-LEAST-ONCE: Retry until acknowledged
  - Message may be delivered multiple times
  - Consumer must handle duplicates (idempotency)
  - Use for: most production systems with idempotent consumers

EXACTLY-ONCE: Each message processed exactly one time
  - Hardest to achieve in distributed systems
  - Requires coordination between source, processor, and sink
  - Use for: financial transactions, inventory updates

Implementing Exactly-Once

Strategy 1: Idempotent Consumer (most practical)
  - Assign each message a unique ID at the producer
  - Consumer checks if ID was already processed before acting
  - Store processed IDs in a deduplication table with TTL

Strategy 2: Transactional Outbox
  - Write business data and outbox event in same DB transaction
  - Separate process reads outbox and publishes to message broker
  - Guarantees: if business data committed, event will be published

Strategy 3: Kafka Transactions (Kafka-specific)
  - Producer uses transactional API
  - Read-process-write cycle wrapped in Kafka transaction
  - Consumer reads only committed messages (read_committed isolation)

Idempotent Consumer Pattern

# Idempotent consumer with deduplication
class IdempotentConsumer:
    def __init__(self, db, processor):
        self.db = db
        self.processor = processor

    def handle_message(self, message):
        message_id = message.headers.get('idempotency-key')

        # Check if already processed
        if self.db.exists('processed_messages', message_id):
            return  # Skip duplicate

        # Process within a transaction
        with self.db.transaction() as tx:
            # Record that we processed this message
            tx.insert('processed_messages', {
                'message_id': message_id,
                'processed_at': datetime.utcnow(),
                'expires_at': datetime.utcnow() + timedelta(days=7)
            })

            # Execute business logic
            self.processor.process(message.value, tx)

        # Commit consumer offset only after successful processing
        message.commit()

Transactional Outbox Pattern

-- Business table and outbox in same database
CREATE TABLE orders (
    id UUID PRIMARY KEY,
    customer_id UUID NOT NULL,
    total_amount DECIMAL(10,2),
    status TEXT DEFAULT 'pending',
    created_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE outbox (
    id BIGSERIAL PRIMARY KEY,
    aggregate_type TEXT NOT NULL,     -- 'order'
    aggregate_id UUID NOT NULL,       -- order.id
    event_type TEXT NOT NULL,         -- 'order_created'
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT now(),
    published_at TIMESTAMPTZ          -- NULL until published
);

-- Application code: single transaction
BEGIN;
  INSERT INTO orders (id, customer_id, total_amount, status)
  VALUES ('abc-123', 'user-456', 99.99, 'pending');

  INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
  VALUES ('order', 'abc-123', 'order_created',
          '{"order_id": "abc-123", "customer_id": "user-456", "total": 99.99}');
COMMIT;

-- Outbox publisher (separate process, polls or uses CDC):
-- 1. SELECT * FROM outbox WHERE published_at IS NULL ORDER BY id LIMIT 100;
-- 2. Publish each event to Kafka/message broker
-- 3. UPDATE outbox SET published_at = now() WHERE id IN (...);

Backpressure

What Is Backpressure

Backpressure is a mechanism for a slow consumer to signal upstream
producers to slow down, preventing buffer overflow and cascading failures.

Without backpressure:
  Producer (1000 msg/s) → Buffer (fills up) → Consumer (500 msg/s)
  Result: Buffer overflow → OOM → data loss or crash

With backpressure:
  Producer (500 msg/s) ← "slow down" ← Buffer → Consumer (500 msg/s)
  Result: System operates at consumer's pace, no data loss

Backpressure Strategies

1. BLOCKING (synchronous backpressure)
   Producer blocks when buffer is full
   Simple, but can cause cascading slowdowns upstream

2. DROPPING (load shedding)
   Drop oldest or newest messages when buffer exceeds threshold
   Use for: metrics, sensor data where latest value matters most

3. BUFFERING with bounds
   Bounded queue with configurable overflow behavior
   Monitor buffer size as a key health metric

4. RATE LIMITING at the source
   Token bucket or leaky bucket at producer
   Consumer advertises its capacity, producer respects it

5. DYNAMIC SCALING
   Auto-scale consumer count based on queue depth
   Lag-based scaling: if consumer lag > threshold, add consumers

Implementation Examples

# Kafka consumer with backpressure via partition pause/resume
from confluent_kafka import Consumer

class BackpressureConsumer:
    def __init__(self, config, max_buffered=1000):
        self.consumer = Consumer(config)
        self.max_buffered = max_buffered
        self.buffer = []
        self.paused_partitions = set()

    def poll_loop(self):
        while True:
            # If buffer is too full, pause consuming
            if len(self.buffer) >= self.max_buffered:
                assignments = self.consumer.assignment()
                unpaused = [tp for tp in assignments
                           if tp not in self.paused_partitions]
                if unpaused:
                    self.consumer.pause(unpaused)
                    self.paused_partitions.update(unpaused)

            # If buffer has drained, resume consuming
            elif len(self.buffer) < self.max_buffered * 0.5:
                if self.paused_partitions:
                    self.consumer.resume(list(self.paused_partitions))
                    self.paused_partitions.clear()

            msg = self.consumer.poll(timeout=0.1)
            if msg and not msg.error():
                self.buffer.append(msg)

            # Process buffered messages
            self.process_batch()

    def process_batch(self):
        batch = self.buffer[:100]
        self.buffer = self.buffer[100:]
        for msg in batch:
            self.process(msg)

// Reactive Streams backpressure (Project Reactor / RxJava)
// Consumer requests only what it can handle

Flux.from(dataSource)
    .onBackpressureBuffer(1000)          // Buffer up to 1000 items
    .onBackpressureDrop(dropped -> .// Drop if buffer full
        metrics.increment("backpressure.dropped"))
    .flatMap(item -> processAsync(item),
             /* concurrency */ 10)        // Max 10 concurrent
    .subscribe();

Change Data Capture (CDC)

CDC Architecture

┌──────────┐    ┌───────────┐    ┌──────────┐    ┌───────────┐
│ Source DB │───>│ CDC Tool  │───>│ Kafka    │───>│ Consumers │
│ (Postgres)│   │ (Debezium)│    │ Topics   │    │           │
│           │   │           │    │          │    │ - Search  │
│ WAL/Binlog│   │ Reads log │    │ Events:  │    │ - Cache   │
│           │   │ changes   │    │ insert   │    │ - Analytics│
└──────────┘    └───────────┘    │ update   │    │ - Replica │
                                 │ delete   │    └───────────┘
                                 └──────────┘

Debezium Configuration

{
  "name": "postgres-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "db.example.com",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${secrets:db-password}",
    "database.dbname": "myapp",
    "topic.prefix": "myapp",
    "table.include.list": "public.orders,public.customers",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "debezium_pub",

    "transforms": "route",
    "transforms.route.type": "io.debezium.transforms.ByLogicalTableRouter",
    "transforms.route.topic.regex": "(.*)\\.(.*)",
    "transforms.route.topic.replacement": "$1.$2",

    "snapshot.mode": "initial",
    "tombstones.on.delete": true,
    "decimal.handling.mode": "string",
    "time.precision.mode": "connect"
  }
}

CDC Event Structure

{
  "before": {
    "id": 1001,
    "status": "pending",
    "total": "99.99"
  },
  "after": {
    "id": 1001,
    "status": "shipped",
    "total": "99.99"
  },
  "source": {
    "version": "2.5.0",
    "connector": "postgresql",
    "name": "myapp",
    "ts_ms": 1705000000000,
    "db": "myapp",
    "schema": "public",
    "table": "orders",
    "lsn": 123456789
  },
  "op": "u",
  "ts_ms": 1705000000500
}

Data Pipeline Patterns

Fan-Out / Fan-In

Fan-Out: One source, multiple consumers with different processing

                    ┌─→ Consumer A (search indexing)
Source → Topic ─────┼─→ Consumer B (analytics aggregation)
                    ├─→ Consumer C (notification triggers)
                    └─→ Consumer D (cache warming)

Each consumer has its own consumer group and offset.
Consumers process at their own pace independently.

Fan-In: Multiple sources converge into one processor

Source A (orders) ──┐
Source B (payments)──┼─→ Enrichment Processor → Unified View
Source C (shipping)──┘

Requires: event correlation by key (order_id), handling out-of-order
arrival, and windowed joins.

Dead Letter Queue (DLQ)

Main Topic → Consumer → Success → Output
                │
                └─→ Failure (after N retries) → Dead Letter Topic
                                                      │
                                                      ↓
                                                DLQ Consumer
                                                (alerting +
                                                 manual review)

Configuration:
  Max retries: 3
  Retry delay: exponential (1s, 5s, 25s)
  DLQ retention: 14 days
  Alert: if DLQ depth > 100 messages in 5 minutes

DLQ message enrichment:
  - Original message payload
  - Error message and stack trace
  - Retry count and timestamps
  - Source topic and partition
  - Consumer group that failed

Data Processing Guarantees Comparison

Framework         | Delivery  | Stateful | Latency    | Throughput
------------------|-----------|----------|------------|------------
Kafka Streams     | Exactly-once (Kafka-to-Kafka) | Yes | ms | High
Apache Flink      | Exactly-once (checkpoints) | Yes | ms | Very High
Spark Structured  | Exactly-once (checkpoints) | Yes | 100ms+ | Very High
Apache Beam       | Exactly-once (runner-dependent) | Yes | Varies | High
AWS Kinesis       | At-least-once | Limited | 200ms+ | Moderate
Pub/Sub + Dataflow| Exactly-once | Yes | 100ms+ | High

Choosing a framework:
  Kafka-native, JVM: Kafka Streams (simplest for Kafka-to-Kafka)
  Complex event processing, low latency: Apache Flink
  Batch + stream unified, large scale: Spark Structured Streaming
  Multi-cloud portability: Apache Beam
  AWS-native, managed: Kinesis + Lambda
  GCP-native, managed: Pub/Sub + Dataflow

Data Pipeline Monitoring

Key Metrics

Consumer Lag:
  - Messages in topic minus consumer offset
  - Alert if lag > N messages or growing for > M minutes
  - Critical: if lag exceeds retention period, data loss occurs

Processing Throughput:
  - Messages processed per second per consumer
  - Compare to ingestion rate to detect falling behind

Processing Latency:
  - Time from message production to consumer processing complete
  - Track p50, p95, p99 separately

Error Rate:
  - Failed messages / total messages
  - Alert at >0.1% error rate
  - Track by error type for root cause analysis

Checkpoint Duration (Flink/Spark):
  - Time to complete a state checkpoint
  - If checkpoint duration > checkpoint interval, system is unstable

Data-Intensive Architecture Checklist

Design:
[ ] Chosen batch vs stream vs hybrid based on latency requirements
[ ] Defined delivery guarantees (at-least-once + idempotency in most cases)
[ ] Designed schema evolution strategy (Avro schemas with registry)
[ ] Planned for data replay and reprocessing capability
[ ] Identified stateful operations and state management approach

Implementation:
[ ] Idempotent consumers with deduplication for at-least-once processing
[ ] Dead letter queues for failed message handling
[ ] Backpressure mechanism to prevent consumer overload
[ ] Schema registry for message format compatibility
[ ] Transactional outbox for reliable event publishing from databases

Operations:
[ ] Consumer lag monitoring with alerting thresholds
[ ] Processing throughput and latency dashboards
[ ] Error rate tracking with automatic DLQ alerting
[ ] Capacity planning for topic partitions and consumer scaling
[ ] Retention policies aligned with reprocessing requirements
[ ] Runbook for common failure modes (consumer restart, rebalance, lag spike)

When to Use

Use this skill when:

• Designing or implementing data intensive designer solutions
• Reviewing or improving existing data intensive designer approaches
• Making architectural or implementation decisions about data intensive designer
• Learning data intensive designer patterns and best practices
• Troubleshooting data intensive designer-related issues

Do NOT use this skill when:

• The question is about a fundamentally different technology domain
• A more specific sibling skill covers the exact topic needed
• The user needs a complete hands-on tutorial rather than expert guidance

Output Format

# Data Intensive Designer Analysis

## Context Assessment
[Situation summary and constraints]

## Recommended Approach
[Primary recommendation with rationale]

## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]

## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]

## Next Steps
- [Immediate action item]
- [Follow-up action item]

Example

Input: "Help me implement data intensive designer for a medium-scale production application"

Output: A structured analysis covering current state assessment, recommended data intensive designer approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.

Edge Cases

• Legacy system integration: When data intensive designer must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
• Scale mismatch: When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
• Team skill gaps: When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
• Conflicting requirements: When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities