Apache Flink - Advanced Stream Processing

Spark Streaming Model:

Time:     0s         1s         2s         3s         4s
          │          │          │          │          │
Events:   ████       ████       ████       ████       ████
          Batch 1    Batch 2    Batch 3    Batch 4    Batch 5
          │          │          │          │          │
          └──────────┴──────────┴──────────┴──────────┘

          Process each micro-batch using Spark RDDs

Latency:  ~500ms - 2 seconds (limited by batch interval)
Model:    Discrete time intervals, not continuous
State:    Managed at batch boundaries

from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col

spark = SparkSession.builder.appName("SparkStreaming").getOrCreate()

# Read stream from Kafka (micro-batches)
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "transactions") \
    .load()

# Parse JSON
transactions = df.selectExpr("CAST(value AS STRING)") \
    .selectExpr("from_json(value, schema) as data") \
    .select("data.*")

# Windowed aggregation (processing time by default)
windowed_totals = transactions \
    .groupBy(
        window("timestamp", "1 minute"),  # 1-minute windows
        "user_id"
    ) \
    .agg({"amount": "sum"})

# Write to console (triggers micro-batch processing)
query = windowed_totals.writeStream \
    .outputMode("update") \
    .format("console") \
    .trigger(processingTime="5 seconds")  # Process every 5 seconds \
    .start()

query.awaitTermination()

Flink Model:

Time:     0s         1s         2s         3s         4s
          │          │          │          │          │
Events:   █─█──█───█─█─█──█───█──█─█────█──█───█─█──

          Each event processed immediately as it arrives

Latency:  ~1ms - 100ms (sub-second)
Model:    Continuous, record-at-a-time processing
State:    Maintained continuously with checkpoints

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common import Time, WatermarkStrategy
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream.connectors.kafka import KafkaSource

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)

# Read stream from Kafka (true streaming, not batches)
kafka_source = KafkaSource.builder() \
    .set_bootstrap_servers("localhost:9092") \
    .set_topics("transactions") \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .build()

# Define watermark strategy for event time
watermark_strategy = WatermarkStrategy \
    .for_bounded_out_of_orderness(Time.seconds(5)) \  # 5 sec max lateness
    .with_timestamp_assigner(lambda event: event['timestamp'])

# Create stream with event time
stream = env.from_source(
    kafka_source,
    watermark_strategy,
    "kafka-source"
)

# Parse JSON and extract fields
transactions = stream.map(lambda x: json.loads(x))

# Windowed aggregation using EVENT TIME (not processing time!)
windowed_totals = transactions \
    .key_by(lambda x: x['user_id']) \
    .window(TumblingEventTimeWindows.of(Time.minutes(1))) \  # 1-min event time windows
    .reduce(lambda a, b: {
        'user_id': a['user_id'],
        'total': a['total'] + b['amount']
    })

# Sink to console
windowed_totals.print()

# Execute (streaming job runs continuously)
env.execute("Flink Streaming")

PROCESSING TIME (arrival time):
Events arrive:    E1(10:00) → E2(10:01) → E3(10:02)
Processed at:     10:05       10:06       10:07
Window:           [10:05-10:06], [10:06-10:07], [10:07-10:08]

Problem: E1 happened at 10:00 but gets windowed at 10:05!
         Results wrong if events delayed or out-of-order


EVENT TIME (when event actually occurred):
Events arrive:    E1(10:00) → E3(10:02) → E2(10:01)  ← Out of order!
Event timestamp:  10:00       10:02       10:01
Window:           [10:00-10:01], [10:01-10:02], [10:02-10:03]

Correct: Each event placed in window based on its actual timestamp
         Results accurate regardless of arrival order or delays

Watermark Mechanics:

Time:        10:00   10:01   10:02   10:03   10:04   10:05
             │       │       │       │       │       │
Events:      E1      E3      E2      E5      E4      │
             ts:00   ts:02   ts:01   ts:04   ts:03   │
                                                      │
Watermark:   00      01      02      03      04      │
             │       │       └─Window[00-01] closes  │
             │       │         (E1, E2 included)     │
             │       │                                │
             │       └─Window[01-02] closes          │
             │         (E2, E3 included)             │
             │                                        │
             └─E1 processed immediately               │

Watermark = max(event_time) - allowed_lateness
          = 10:02 - 1 min = 10:01

Window [10:00-10:01] closes when watermark reaches 10:01

from pyflink.common import WatermarkStrategy, Time

# Strategy 1: Bounded out-of-orderness
# Assumes events can be up to 5 seconds late
watermark_strategy = WatermarkStrategy \
    .for_bounded_out_of_orderness(Time.seconds(5)) \
    .with_timestamp_assigner(lambda event: event['timestamp'])

# Strategy 2: Monotonous timestamps
# Assumes events always arrive in order (no lateness)
watermark_strategy = WatermarkStrategy \
    .for_monotonous_timestamps() \
    .with_timestamp_assigner(lambda event: event['timestamp'])

# Strategy 3: Custom watermark generator
class CustomWatermarkGenerator(WatermarkGenerator):
    def __init__(self):
        self.max_timestamp = 0

    def on_event(self, event, timestamp):
        # Track maximum timestamp seen
        self.max_timestamp = max(self.max_timestamp, timestamp)

    def on_periodic_emit(self):
        # Emit watermark: max_timestamp - 10 seconds
        return Watermark(self.max_timestamp - 10000)

# Apply watermark to stream
stream = env.from_source(kafka_source, watermark_strategy, "source")

from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common import Time
from pyflink.datastream import OutputTag

# Define late data tag
late_data_tag = OutputTag("late-data", Types.STRING())

# Configure window with allowed lateness
windowed_stream = stream \
    .key_by(lambda x: x['user_id']) \
    .window(TumblingEventTimeWindows.of(Time.minutes(1))) \
    .allowed_lateness(Time.minutes(5)) \  # Allow 5 min late data
    .side_output_late_data(late_data_tag) \  # Send late data to side output
    .reduce(lambda a, b: {
        'user_id': a['user_id'],
        'total': a['total'] + b['amount']
    })

# Get late events from side output
late_events = windowed_stream.get_side_output(late_data_tag)

# Process late events separately (log, store for reprocessing, etc.)
late_events.print("LATE EVENT")

from pyflink.datastream import StreamExecutionEnvironment, KeyedProcessFunction
from pyflink.common.typeinfo import Types
from pyflink.datastream.state import ValueStateDescriptor

class FraudDetector(KeyedProcessFunction):
    """Detect fraud: alert if 2 large transactions within 1 minute"""

    def open(self, runtime_context):
        # Initialize state: last large transaction timestamp
        self.last_large_tx_state = runtime_context.get_state(
            ValueStateDescriptor("last-large-tx", Types.LONG())
        )

    def process_element(self, transaction, ctx):
        # Check if transaction is large (> $1000)
        if transaction['amount'] > 1000:
            # Get last large transaction time
            last_large_tx_time = self.last_large_tx_state.value()

            if last_large_tx_time is not None:
                # Check if within 1 minute
                time_diff = transaction['timestamp'] - last_large_tx_time
                if time_diff < 60000:  # 60 seconds in milliseconds
                    # FRAUD DETECTED!
                    yield {
                        'user_id': transaction['user_id'],
                        'alert': 'FRAUD',
                        'reason': f'Two large transactions within {time_diff/1000}s',
                        'amount1': last_large_tx_time,
                        'amount2': transaction['amount']
                    }

            # Update state with current transaction time
            self.last_large_tx_state.update(transaction['timestamp'])

# Apply fraud detector
env = StreamExecutionEnvironment.get_execution_environment()

transactions = env.from_collection([
    {'user_id': 'U123', 'amount': 1500, 'timestamp': 1000},
    {'user_id': 'U123', 'amount': 2000, 'timestamp': 30000},  # 29s later - FRAUD!
    {'user_id': 'U456', 'amount': 500, 'timestamp': 5000},
    {'user_id': 'U123', 'amount': 3000, 'timestamp': 120000}, # >1min later - OK
])

alerts = transactions \
    .key_by(lambda x: x['user_id']) \
    .process(FraudDetector())

alerts.print()
env.execute("Fraud Detection")

from pyflink.datastream import StreamExecutionEnvironment

env = StreamExecutionEnvironment.get_execution_environment()

# Configure RocksDB state backend (for large state)
env.set_state_backend(
    "rocksdb",
    checkpoint_path="s3://my-bucket/checkpoints"
)

# Enable checkpointing every 60 seconds
env.enable_checkpointing(60000)  # milliseconds

# Create a savepoint manually (via CLI)
$ flink savepoint <job-id> s3://my-bucket/savepoints

# Stop job with savepoint
$ flink stop --savepointPath s3://my-bucket/savepoints <job-id>

# Restore from savepoint (e.g., after code upgrade)
$ flink run -s s3://my-bucket/savepoints/savepoint-123 my-job.jar

Flink's Exactly-Once Mechanism:

1. Distributed Snapshots (Chandy-Lamport Algorithm):
   ┌─────────────────────────────────────────────┐
   │  Checkpoint Barrier flows through pipeline  │
   │                                             │
   │  Source ─barrier→ Op1 ─barrier→ Op2 ─Sink   │
   │    │               │         │              │
   │    └─ state ───────┴─ state ┴──── state     │
   │                                             │
   │  When barrier reaches operator:             │
   │    • Snapshot state                         │
   │    • Forward barrier                        │
   └─────────────────────────────────────────────┘

2. Two-Phase Commit for Sinks:
   ┌─────────────────────────────────────────────┐
   │  Checkpoint N started:                      │
   │    Phase 1 (Pre-commit): Stage writes       │
   │                                             │
   │  Checkpoint N completes:                    │
   │    Phase 2 (Commit): Atomically commit      │
   └─────────────────────────────────────────────┘

3. On Failure:
   • Restore state from last successful checkpoint
   • Replay events from checkpoint onwards
   • No duplicates or data loss

from pyflink.datastream import StreamExecutionEnvironment, CheckpointingMode

env = StreamExecutionEnvironment.get_execution_environment()

# Enable checkpointing with exactly-once mode
env.enable_checkpointing(60000, CheckpointingMode.EXACTLY_ONCE)

# Configure checkpoint settings
checkpoint_config = env.get_checkpoint_config()

# Checkpoint timeout (max time for checkpoint to complete)
checkpoint_config.set_checkpoint_timeout(300000)  # 5 minutes

# Minimum time between checkpoints (avoid too frequent)
checkpoint_config.set_min_pause_between_checkpoints(30000)  # 30 seconds

# Max concurrent checkpoints
checkpoint_config.set_max_concurrent_checkpoints(1)

# Retain checkpoints on job cancellation (for recovery)
checkpoint_config.enable_externalized_checkpoints(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
)

# Configure Kafka sink with exactly-once (requires Kafka 0.11+)
from pyflink.datastream.connectors.kafka import KafkaSink, KafkaRecordSerializationSchema

kafka_sink = KafkaSink.builder() \
    .set_bootstrap_servers("localhost:9092") \
    .set_record_serializer(
        KafkaRecordSerializationSchema.builder()
            .set_topic("output-topic")
            .set_value_serialization_schema(SimpleStringSchema())
            .build()
    ) \
    .set_delivery_guarantee(DeliveryGuarantee.EXACTLY_ONCE) \  # Exactly-once!
    .set_transactional_id_prefix("flink-txn-") \
    .build()

# Write to Kafka with exactly-once guarantees
stream.sink_to(kafka_sink)

env.execute("Exactly-Once Pipeline")