1298  Handling Stream Processing Challenges

Part of: Stream Processing for IoT

Previous: Building IoT Streaming Pipelines

1298.1 Handling Real-World Challenges

⏱️ ~20 min | ⭐⭐⭐ Advanced | 📋 P10.C14.U05

Production IoT stream processing systems face several significant challenges that require careful consideration in system design.

1298.1.1 Late Data and Watermarks

In distributed systems, events frequently arrive out of order or late due to network delays, device buffering, or connectivity issues. Watermarks provide a mechanism to handle this.

Mermaid diagram

Figure 1298.1: Watermark mechanism showing how late-arriving events are handled relative to window boundaries and watermark progression

Watermark Definition: A watermark is a timestamp that indicates the system believes no events with earlier timestamps will arrive.

Flink Watermark Configuration:

from pyflink.common import WatermarkStrategy, Duration

# Allow events up to 2 minutes late
watermark_strategy = (
    WatermarkStrategy
    .for_bounded_out_of_orderness(Duration.of_minutes(2))
    .with_timestamp_assigner(
        lambda event, timestamp: event['event_time']
    )
)

stream.assign_timestamps_and_watermarks(watermark_strategy)

This processes events up to 2 minutes late, updating already-emitted results.

{
  const container = document.getElementById('kc-data-6');
  if (container && typeof InlineKnowledgeCheck !== 'undefined') {
    container.innerHTML = '';
    container.appendChild(InlineKnowledgeCheck.create({
      question: "A remote weather station in a mountainous area loses satellite connectivity for 2 hours, buffering readings locally. When connectivity is restored, it sends 720 readings (6 per minute) with timestamps from the offline period. The stream processor has 5-minute tumbling windows with a 30-second watermark delay. What happens to the buffered readings?",
      options: [
        {text: "All 720 readings are processed correctly into their respective 5-minute windows", correct: false, feedback: "With only 30-second watermark delay, windows from 2 hours ago have long since closed. The readings arrive far too late to be included in their original windows."},
        {text: "The readings are dropped silently because they exceed the watermark delay", correct: false, feedback: "If dropped silently, this would violate the 'never silently lose data' principle. While they may not fit into original windows, proper stream processing shouldn't discard them without notification."},
        {text: "Readings are routed to a side output for late data, processed separately or in batch", correct: true, feedback: "Correct! Best practice is to configure 'allowed lateness' with side outputs for very late data. The 2-hour-late readings exceed any reasonable watermark grace period, so they go to a side output topic/queue for batch reprocessing or manual review rather than being silently dropped."},
        {text: "The watermark automatically adjusts to accommodate the 2-hour delay", correct: false, feedback: "Watermarks don't automatically adjust to extreme delays - this would defeat their purpose of bounding latency. If watermarks waited 2 hours for stragglers, real-time processing would be impossible."}
      ],
      difficulty: "hard",
      topic: "data-processing"
    }));
  }
}

CautionPitfall: Window Size Mismatch with Business Requirements

The Mistake: Choosing window sizes based on technical convenience (nice round numbers like 1 minute or 5 minutes) rather than aligning with actual business or physical process requirements, resulting in analytics that miss critical patterns.

Why It Happens: Default configurations in tutorials use convenient time intervals. Engineers optimize for system performance without consulting domain experts. The relationship between window size and signal detection thresholds is not analyzed mathematically. Copy-pasting configurations between different IoT deployments.

The Fix: Derive window size from the physics of what you’re monitoring. For equipment anomaly detection, window size should be 2-3x the expected anomaly duration (a bearing failure signature lasting 2 seconds needs windows of 4-6 seconds minimum). For process control, align windows with process cycle times. For user behavior analytics, use session windows that naturally capture activity bursts. Always validate that chosen window sizes capture at least 10 samples of the fastest phenomenon you need to detect.

TipUnderstanding Exactly-Once Processing

Core Concept: Exactly-once semantics guarantee that each event in a stream is processed precisely one time, even when failures, retries, or network partitions occur.

Why It Matters: In IoT systems handling billing, safety alerts, or inventory updates, processing an event twice causes incorrect totals, duplicate charges, or redundant emergency responses. Processing zero times means lost revenue, missed alarms, or phantom inventory. A smart meter billing system that double-counts readings will overcharge customers; one that drops readings will undercharge and lose revenue.

Key Takeaway: Exactly-once requires three coordinated mechanisms: idempotent operations (same input always produces same output), transactional commits (offset and result saved atomically), and deterministic processing (no random or time-dependent logic). Accept the 10-30% latency overhead as the cost of correctness for critical data paths.

1298.1.2 Exactly-Once Processing

For IoT applications involving billing, safety alerts, or financial transactions, processing each event exactly once is critical.

Problem: In distributed systems, failures can cause: - Lost data: Event processed but result not saved before crash - Duplicates: Event processed twice after retry

Solution: Exactly-once semantics through:

1. Idempotent operations: Processing same event twice produces same result
2. Transactional writes: Atomically commit offsets and results
3. Distributed snapshots: Checkpoint entire pipeline state consistently

Kafka Example: Kafka Streams achieves exactly-once by: - Reading offsets from source topics - Processing events - Writing results to output topics - Committing all changes in a single transaction

Performance Impact: Exactly-once adds 10-30% latency overhead compared to at-least-once processing, but eliminates duplicate or lost data.

{
  const container = document.getElementById('kc-data-7');
  if (container && typeof InlineKnowledgeCheck !== 'undefined') {
    container.innerHTML = '';
    container.appendChild(InlineKnowledgeCheck.create({
      question: "A smart meter billing system processes electricity consumption events to calculate monthly charges. The system uses at-least-once delivery with automatic retries. During a network hiccup, one consumption event (5 kWh at $0.15/kWh) is delivered twice. Without deduplication, what billing error occurs?",
      options: [
        {text: "Customer is undercharged by $0.75", correct: false, feedback: "At-least-once delivery causes duplicates, not lost messages. The event is processed twice, not zero times, so the customer would be overcharged, not undercharged."},
        {text: "Customer is overcharged by $0.75 (double-counted consumption)", correct: true, feedback: "Correct! The 5 kWh event is processed twice, counting as 10 kWh. At $0.15/kWh, the customer is charged $1.50 instead of the correct $0.75 - an overcharge of $0.75. This is why billing systems require exactly-once semantics or application-level deduplication."},
        {text: "No error - Kafka automatically deduplicates messages", correct: false, feedback: "Kafka provides at-least-once delivery by default, not automatic deduplication. You must enable idempotent producers and implement consumer-side deduplication for exactly-once semantics."},
        {text: "The billing system rejects the duplicate and processes correctly", correct: false, feedback: "Standard at-least-once processing doesn't automatically detect duplicates. Without explicit deduplication logic (checking message IDs or using exactly-once transactions), both messages are processed independently."}
      ],
      difficulty: "medium",
      topic: "data-processing"
    }));
  }
}

WarningTradeoff: Exactly-Once vs At-Least-Once Processing Semantics

Option A: At-Least-Once Processing - Guarantee no data loss, accept possible duplicates - Latency: 5-20ms per event (lower overhead) - Throughput: 200K-500K events/second - Implementation complexity: Low (simple retry logic) - Failure recovery: Fast (just replay from offset) - Risk: Duplicate processing on failures (~0.01-0.1% of events during outages) - Cost: Standard Kafka cluster ~$300-800/month

Option B: Exactly-Once Processing - Guarantee each event processed precisely once - Latency: 20-50ms per event (10-30% overhead for transactions) - Throughput: 100K-300K events/second - Implementation complexity: High (transactional writes, idempotency keys, checkpointing) - Failure recovery: Slower (must restore checkpoint state) - Risk: None for duplicates, but more complex failure modes - Cost: Higher infrastructure + engineering (~$500-1,500/month + 2x development time)

Decision Factors: - Choose At-Least-Once when: Duplicates are tolerable (logging, metrics aggregation), downstream systems are idempotent by design (upserts with unique keys), latency is critical (<10ms SLA), throughput requirements exceed 300K events/second - Choose Exactly-Once when: Financial transactions (billing, payments), safety-critical alerts (fire, gas leak, equipment failure), regulatory compliance requires audit accuracy, downstream systems cannot handle duplicates (counters, inventory) - Hybrid approach: At-least-once delivery with application-level deduplication (Redis SET with TTL) achieves practical exactly-once at lower complexity for many IoT use cases

CautionPitfall: Confusing At-Least-Once with Exactly-Once

The Mistake: Assuming that achieving at-least-once delivery (no data loss) automatically provides exactly-once semantics, leading to duplicate processing that corrupts counts, sums, and triggers redundant alerts.

Why It Happens: “At-least-once” sounds reassuring and is easier to implement than exactly-once. The duplicate problem only manifests under failures, which are rare during testing. Engineers conflate message delivery guarantees (what Kafka provides) with processing guarantees (what your application must implement). Retry mechanisms are added without idempotency checks.

The Fix: For exactly-once processing, implement all three components: (1) Idempotent operations using unique event IDs to detect and skip duplicates, (2) Transactional commits that atomically update both your output and your consumer offset, and (3) Deterministic processing that produces identical outputs for identical inputs. Use Kafka’s transactional API or Flink’s exactly-once checkpointing. For simpler cases, deduplication tables with TTL can filter duplicates at the application layer. Accept the 10-30% latency overhead as the cost of correctness.

CautionPitfall: Timestamp Precision Mismatch Between Systems

The Mistake: Assuming all systems in your streaming pipeline use the same timestamp precision, causing data to land in wrong time windows or get dropped as “too late.”

Why It Happens: Different technologies use different timestamp precisions by default: - Python time.time(): seconds (float, ~1.7 billion) - JavaScript Date.now(): milliseconds (~1.7 trillion) - Java System.currentTimeMillis(): milliseconds - Kafka message timestamps: milliseconds - InfluxDB: nanoseconds by default - TimescaleDB: microseconds (TIMESTAMPTZ)

When a sensor sends 1704672000 (seconds) but Kafka expects milliseconds, the event appears to be from January 1970.

The Fix: Standardize on milliseconds and validate at boundaries:

import time
from datetime import datetime, timezone

def normalize_timestamp(ts, expected_precision='ms'):
    """Convert any timestamp to milliseconds."""
    # Detect precision by magnitude
    if ts < 1e10:  # Seconds (10 digits)
        return int(ts * 1000)
    elif ts < 1e13:  # Milliseconds (13 digits)
        return int(ts)
    elif ts < 1e16:  # Microseconds (16 digits)
        return int(ts / 1000)
    else:  # Nanoseconds (19 digits)
        return int(ts / 1_000_000)

def validate_timestamp(ts_ms, max_future_sec=60, max_past_days=7):
    """Reject timestamps outside reasonable bounds."""
    now_ms = int(time.time() * 1000)
    if ts_ms > now_ms + (max_future_sec * 1000):
        raise ValueError(f"Timestamp {ts_ms} is in the future")
    if ts_ms < now_ms - (max_past_days * 86400 * 1000):
        raise ValueError(f"Timestamp {ts_ms} is too old")
    return ts_ms

Validation rules: Reject timestamps before 2020 (likely precision error), reject timestamps more than 1 minute in the future (clock skew), log warnings for timestamps more than 1 hour old (late data).

TipUnderstanding Backpressure

Core Concept: Backpressure is a flow control mechanism where downstream components signal upstream producers to slow down when they cannot keep up with the incoming data rate.

Why It Matters: Without backpressure handling, fast producers overwhelm slow consumers, causing unbounded queue growth, memory exhaustion, and eventual system crashes. In IoT deployments, this commonly occurs after network outages when thousands of sensors simultaneously reconnect and flood the pipeline with buffered data. A factory with 10,000 sensors coming online simultaneously after a 1-hour outage dumps 36 million accumulated readings in seconds.

Key Takeaway: Design for backpressure from day one using bounded queues (not unbounded), implement producer throttling when buffers fill, and add monitoring for queue depth. The choice is simple: slow down gracefully under load, or crash catastrophically when buffers overflow.

1298.1.3 Backpressure

Backpressure occurs when data arrives faster than your pipeline can process it. This is common in IoT systems during: - Traffic spikes (e.g., all sensors reporting after network outage) - Slow downstream systems (e.g., database overloaded) - Processing bottlenecks (e.g., expensive ML inference)

%%{init: {'theme': 'base', 'themeVariables': {'primaryColor': '#2C3E50', 'primaryTextColor': '#fff', 'primaryBorderColor': '#16A085', 'lineColor': '#E67E22', 'secondaryColor': '#7F8C8D', 'tertiaryColor': '#fff'}}}%%
graph LR
    subgraph "Healthy Pipeline"
        IN1[Input<br/>1000 evt/sec]
        PROC1[Process<br/>1200 evt/sec]
        OUT1[Output<br/>1200 evt/sec]
    end

    subgraph "Backpressure Scenario"
        IN2[Input<br/>5000 evt/sec]
        PROC2[Process<br/>1200 evt/sec<br/>Queue growing!]
        OUT2[Output<br/>500 evt/sec<br/>DB Overloaded]
    end

    IN1 --> PROC1
    PROC1 --> OUT1

    IN2 -.Overwhelms.-> PROC2
    PROC2 -.Bottleneck.-> OUT2

    style IN1 fill:#16A085,stroke:#2C3E50,stroke-width:2px,color:#fff
    style PROC1 fill:#16A085,stroke:#2C3E50,stroke-width:2px,color:#fff
    style OUT1 fill:#16A085,stroke:#2C3E50,stroke-width:2px,color:#fff

    style IN2 fill:#E67E22,stroke:#2C3E50,stroke-width:2px,color:#fff
    style PROC2 fill:#E67E22,stroke:#2C3E50,stroke-width:2px,color:#fff
    style OUT2 fill:#E67E22,stroke:#2C3E50,stroke-width:2px,color:#fff

Figure 1298.2: Healthy Pipeline versus Backpressure Bottleneck Comparison

Mitigation Strategies:

1. Buffering: Queue events in Kafka (days of retention)
2. Scaling: Add more processing instances
3. Rate limiting: Drop or sample events when overloaded
4. Flow control: Slow down producers (Reactive Streams)
5. Priority queuing: Process critical events first

Flink Backpressure: Flink automatically slows down fast sources when downstream operators can’t keep up, propagating pressure through the pipeline.

{
  const container = document.getElementById('kc-data-8');
  if (container && typeof InlineKnowledgeCheck !== 'undefined') {
    container.innerHTML = '';
    container.appendChild(InlineKnowledgeCheck.create({
      question: "A factory's network recovers after a 30-minute outage. 5,000 sensors immediately reconnect and send buffered readings, flooding the pipeline with 500,000 events in 10 seconds (50,000 events/second vs normal 1,000 events/second). The stream processor has an unbounded input queue. What is the MOST LIKELY outcome?",
      options: [
        {text: "The pipeline processes all events with slightly increased latency", correct: false, feedback: "With unbounded queues and 50x normal load sustained for 10 seconds, the queue grows to 500,000 events. If the processor handles 1,000/sec, the queue takes 500 seconds (8+ minutes) to drain - that's not 'slightly increased' latency."},
        {text: "Events are automatically sampled to maintain throughput", correct: false, feedback: "Unbounded queues don't automatically sample - they just grow indefinitely. Sampling requires explicit configuration with bounded buffers and overflow policies."},
        {text: "Memory exhaustion crash as the unbounded queue consumes all available RAM", correct: true, feedback: "Correct! Unbounded queues are a classic backpressure anti-pattern. At 50,000 events/sec with 1KB each, the queue grows by 50MB/sec. Within minutes, it consumes all heap memory, causing OutOfMemoryError. This is why bounded queues with backpressure signaling are essential."},
        {text: "Kafka automatically rate-limits the reconnecting sensors", correct: false, feedback: "Kafka doesn't rate-limit producers by default - that's the producer's responsibility. The sensors would successfully publish all buffered messages, overwhelming downstream consumers."}
      ],
      difficulty: "hard",
      topic: "data-processing"
    }));
  }
}

WarningTradeoff: Frequent vs Infrequent Checkpointing for Stream Fault Tolerance

Option A (Frequent Checkpoints - Every 10-30 seconds): - Recovery time (RTO): 10-30 seconds (minimal replay from last checkpoint) - Data loss on failure (RPO): 10-30 seconds of in-flight events - Checkpoint overhead: 5-15% of processing capacity consumed by snapshotting - State store I/O: High - frequent writes to S3/HDFS/RocksDB - Barrier alignment latency: 50-200ms added to event latency during checkpoint - Storage cost: Higher - more checkpoint versions retained (e.g., 3 retained = 3x state size) - Recovery predictability: High - consistent, fast recovery regardless of when failure occurs - Best for: Financial transactions, safety alerts, regulatory audit trails

Option B (Infrequent Checkpoints - Every 5-15 minutes): - Recovery time (RTO): 5-15 minutes (significant replay required from Kafka offsets) - Data loss on failure (RPO): 0 with exactly-once (replay from source), 5-15 minutes with at-most-once - Checkpoint overhead: 1-3% of processing capacity - State store I/O: Low - infrequent bulk writes - Barrier alignment latency: Minimal impact on steady-state latency - Storage cost: Lower - fewer checkpoint versions - Recovery predictability: Variable - recovery time depends on when in checkpoint interval failure occurs - Best for: Analytics aggregations, ML feature pipelines, telemetry dashboards

Decision Factors: - Choose Frequent when: SLA requires <1 minute recovery time, processing safety-critical events (equipment failures, security breaches), state size is manageable (<10 GB per task), throughput headroom exists for checkpoint overhead - Choose Infrequent when: Recovery time of 5-15 minutes is acceptable (batch-like analytics), state size is large (>100 GB) making frequent snapshots expensive, maximizing throughput is priority over recovery speed, events can be replayed from durable source (Kafka with multi-day retention) - Adaptive approach: Use incremental checkpointing (Flink RocksDB backend) to get frequent checkpoint benefits with lower I/O overhead - only changed state is written, reducing checkpoint size by 80-95% for slowly-changing state

WarningTradeoff: Hot Standby vs Cold Recovery for Stream Processing Pipelines

Option A (Hot Standby - Active Secondary Pipeline): - Failover time: 1-5 seconds (standby already running, just promote to primary) - RPO: 0-100ms (standby processes same stream in parallel, nearly identical state) - Infrastructure cost: 2x compute resources (both pipelines running continuously) - Operational complexity: High - must manage dual pipelines, output deduplication - State consistency: Requires careful coordination to prevent duplicate outputs - Network cost: 2x ingress if both consume from same Kafka cluster - Availability: 99.99%+ achievable (sub-second failover eliminates most user-visible outages) - Use cases: Trading systems, real-time bidding, autonomous vehicle coordination

Option B (Cold Recovery - Restart from Checkpoint): - Failover time: 30 seconds - 5 minutes (restore checkpoint, replay from Kafka, warm up state) - RPO: Checkpoint interval (10 seconds - 15 minutes of replay needed) - Infrastructure cost: 1x compute resources (standby capacity only during recovery) - Operational complexity: Lower - single pipeline, standard Flink/Spark recovery - State consistency: Guaranteed by checkpoint/replay mechanism - Network cost: 1x ingress during normal operation - Availability: 99.9% typical (minutes of downtime during failures) - Use cases: Analytics dashboards, alerting systems, feature engineering pipelines

Decision Factors: - Choose Hot Standby when: Revenue loss exceeds $1000/minute of downtime, safety-critical systems where 30-second outage is unacceptable, contractual SLA requires 99.99%+ availability, state is too large for fast checkpoint restoration (>1 TB) - Choose Cold Recovery when: Cost optimization is priority (50% infrastructure savings), business tolerates 1-5 minute recovery time, checkpoint-based recovery meets SLA, team lacks expertise for dual-pipeline coordination - Warm standby hybrid: Maintain standby cluster in suspended state (containers allocated but not processing), reducing failover to 10-30 seconds at 1.3x cost instead of 2x - good middle ground for 99.95% availability requirements

CautionPitfall: Timezone-Unaware Window Boundaries

The Mistake: Using UTC-based window boundaries for analytics that will be presented to users in local time, causing daily aggregates to span two calendar days from the user’s perspective.

Why It Happens: Stream processing frameworks default to UTC for window boundaries. A “daily aggregate” window from 00:00:00 UTC to 23:59:59 UTC corresponds to 7:00 PM - 6:59 PM the next day for a user in US Eastern Time. The user sees “Monday’s data” that actually includes Sunday evening and excludes Monday evening.

The Fix: Align window boundaries with business timezone requirements:

# Apache Flink example: timezone-aware daily windows
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common.time import Time
import pytz

# WRONG: UTC-aligned daily windows
stream.window(TumblingEventTimeWindows.of(Time.days(1)))

# CORRECT: Business-timezone-aligned windows
# Option 1: Offset windows by timezone difference
eastern_offset_hours = -5  # EST (adjust for DST)
stream.window(TumblingEventTimeWindows.of(
    Time.days(1),
    Time.hours(eastern_offset_hours)  # Start at midnight Eastern
))

# Option 2: Use timezone-aware key with explicit boundary
def get_local_day_key(event_time_ms, timezone_str='America/New_York'):
    tz = pytz.timezone(timezone_str)
    dt = datetime.fromtimestamp(event_time_ms / 1000, tz=tz)
    return dt.strftime('%Y-%m-%d')  # Local calendar day

DST warning: Timezone offsets change twice a year. A hardcoded offset breaks during Daylight Saving Time transitions. Use timezone libraries that handle DST automatically, or document that your windows use UTC and convert only at display time.

Best practice: Store all timestamps in UTC, aggregate in UTC, and convert to user timezone only in the presentation layer. Document this clearly so analysts don’t misinterpret dashboard data.

{
  const container = document.getElementById('kc-data-9');
  if (container && typeof InlineKnowledgeCheck !== 'undefined') {
    container.innerHTML = '';
    container.appendChild(InlineKnowledgeCheck.create({
      question: "A retail analytics system uses daily tumbling windows (00:00-23:59 UTC) to aggregate store sales. Stores in New York (UTC-5) show that 'Monday sales' include transactions from Sunday 7PM-Monday 6:59PM local time. Executives are confused. What is the ROOT CAUSE of this issue?",
      options: [
        {text: "Network latency causing events to arrive in wrong windows", correct: false, feedback: "Network latency affects processing time, not event time. If using event timestamps, the original transaction time would still be correct - the issue is how window boundaries align with local time."},
        {text: "UTC-based window boundaries don't align with business timezone expectations", correct: true, feedback: "Correct! A 'daily' window from midnight-to-midnight UTC spans 7PM-6:59PM in New York. Executives expect 'Monday sales' to mean local Monday (midnight-midnight ET). The fix is either timezone-aware windows with local boundary offsets, or clear documentation that all reports use UTC with timezone conversion at display time."},
        {text: "Missing data from late-arriving transactions", correct: false, feedback: "Late data would cause undercounting, not timezone misalignment. The issue is that complete data is being grouped into windows that don't match business expectations."},
        {text: "Incorrect clock synchronization between stores and server", correct: false, feedback: "Clock sync issues would cause event times to be wrong, but the pattern described (consistent 5-hour offset) indicates a systematic UTC vs local timezone mismatch, not random clock drift."}
      ],
      difficulty: "medium",
      topic: "data-processing"
    }));
  }
}

TipTradeoff: Stream Processing vs Batch Processing

Decision context: When designing your IoT data pipeline, you must decide whether to process data continuously as it arrives (stream) or collect it first and process in bulk (batch).

Factor	Stream Processing	Batch Processing
Latency	Milliseconds to seconds - insights available immediately	Minutes to hours - must wait for batch window to complete
Cost	Higher per-event cost, continuous compute resources required	Lower per-event cost, efficient bulk processing, spot instances
Accuracy	May miss late-arriving data, approximate aggregates	Complete dataset, exact aggregates, full context available
Complexity	Higher - state management, exactly-once semantics, backpressure	Lower - simpler failure recovery, easier debugging

Choose Stream Processing when:

• Decisions must happen in real-time (fraud detection, safety alerts, autonomous systems)
• Data value degrades rapidly with age (anomaly detection, live monitoring)
• You need continuous, incremental insights rather than periodic reports
• Scale requires processing before storage (filtering noise, aggregating at source)

Choose Batch Processing when:

• Historical analysis and trends matter more than real-time visibility
• Complex joins across large datasets are required (monthly reports, ML training)
• Cost efficiency is paramount and latency tolerance is high
• Data completeness is critical (financial reconciliation, compliance reporting)

Default recommendation: Start with batch processing unless latency requirements are under 1 minute. Many teams prematurely adopt stream processing and struggle with the operational complexity. Consider Lambda Architecture (both) when you need real-time alerts AND accurate historical aggregates.

1298.2 What’s Next

Continue to Common Pitfalls and Worked Examples to learn about real-world pitfalls and see a comprehensive fraud detection pipeline worked example.