Apache Flink Fundamentals

Data Engineering & Modern Data Platforms · Lesson 33 · Big Data Processing and Apache Spark

Lesson Overview

The five-minute tile from Structured Streaming with Spark meets the product promise it was given: minute-scale micro-batches, a ten-minute watermark, a Parquet sink, and a checkpoint Spark owns. A later request asks for something else. Session state per learner has to survive hours, the visible lag on a live classroom tile has to stay under a couple of seconds, and a downstream topic should expose each closed window once to consumers that read committed Kafka records. Those are runtime requirements, not adjectives. Spark can still do a surprising amount of them. Flink is the engine whose default model is a continuous operator graph, event-time watermarks, keyed state, and checkpoints that sinks can join.

This lesson is the comparison and the minimum Flink you need in order to make that choice honestly. Flink runs a JobManager, which coordinates the job and the checkpoint, and TaskManagers, which run the operators. A checkpoint is a barrier that flows through the graph with the records, so that state and source positions form one snapshot. That snapshot gives Flink exactly-once state updates when the source can rewind. End-to-end exactly-once still depends on the sink. The Kafka sink's DeliveryGuarantee.EXACTLY_ONCE is a real connector setting: it requires checkpointing and a transactional id prefix, and consumers see one result only if they read committed records. AT_LEAST_ONCE waits for producer acknowledgements and can still duplicate after a restart. NONE does not wait. The connector page is explicit that offset commits to Kafka are for monitoring. Recovery uses the checkpoint, the same separation Structured Streaming taught, on a different engine.

The examples use Apache Flink's documented surfaces: configuration keys from the checkpointing documentation, the PyFlink KafkaSource and KafkaSink builders, and Flink SQL for the windowed tile because a custom WatermarkStrategy that parses the payload is marked unsupported in PyFlink on the Kafka connector page. Nothing here is a TechLearningPro framework. Where a managed Flink service renames a screen, the JobManager, the checkpoint directory, and the delivery guarantee are still the concepts you operate.

Learning Objectives

By the end of this lesson you will be able to:

1. Name the JobManager and the TaskManagers and say which one coordinates checkpoints.
2. Describe an aligned checkpoint as a barrier aligned across inputs, and say what an unaligned checkpoint changes under backpressure.
3. Separate exactly-once checkpointing of operator state from end-to-end exactly-once delivery.
4. Configure enable_checkpointing, a durable checkpoint directory, and externalized checkpoint retention using documented PyFlink methods.
5. Explain why the Kafka source's committed offsets are not Flink's recovery cursor.
6. Choose DeliveryGuarantee.NONE, AT_LEAST_ONCE, or EXACTLY_ONCE for the Kafka sink, including the transactional id prefix and isolation.level.
7. Place a watermark on event time in Flink SQL and define a tumbling window without inventing a Python watermark API the connector does not provide.
8. Contrast HashMap state and RocksDB state, and name when incremental checkpoints matter.
9. Distinguish a checkpoint from a savepoint.
10. Decide when the Spark micro-batch tile remains the right system, and when Flink is the one to propose.

Prerequisites

• Structured Streaming with Spark: micro-batches, watermarks, checkpoints, and the learning.progress contract. This lesson reuses that contract.
• Spark Joins and Data Shuffling: hot keys still exist. In Flink they become hot keyed-state partitions.
• Distributed Processing Reliability is the next lesson. You can read this one first. The reliability lesson generalizes the checkpoint story across both engines.
• PyFlink matching the Flink version you install. The method names below are from the Flink checkpointing page and the Kafka connector page. Confirm imports against that version's PyFlink API if a class moves between 1.19, 1.20, and 2.x.
• Module 4 still owns Kafka as a cluster. This lesson only uses the connector options those pages publish.

Detailed Core Concepts

What Flink is executing

A Flink job is a graph of operators. Sources emit records. Maps and filters transform them. Keyed operators partition by a key and may hold state for that key. Sinks write out. The client submits the graph. The JobManager turns it into tasks and places those tasks on TaskManagers. Each task is one parallel instance of an operator. Parallelism on the Kafka source is limited by the topic's partitions: extra tasks sit idle, and idle inputs hold watermarks back unless idleness is configured. The connector page states that the Kafka source does not become idle automatically when parallelism exceeds the partition count.

The JobManager hosts the checkpoint coordinator. TaskManagers hold the state and the records. If you remember only one operational picture, remember that: a dead TaskManager loses in-flight records and recovers them from the last checkpoint plus the source; a dead JobManager has to come back, on a highly available setup, before the job can coordinate the next checkpoint. This course does not invent a deployment topology. It tells you to run the JobManager highly available and to put checkpoints on a filesystem both roles can read, which is the checkpointing page's requirement.

Checkpoints and barriers

Checkpointing is off until you turn it on. enable_checkpointing(interval_ms) starts a checkpoint on that interval. The mode defaults to exactly-once. In exactly-once checkpointing, Flink injects a barrier on each input. An aligned checkpoint waits until a task has received the barrier from every input, then snapshots state. Records after the barrier belong to the next checkpoint. The snapshot plus the source positions is one logical instant. On restore, tasks load that state and sources rewind. The result matches a failure-free execution for operator state, which is the sentence the checkpointing documentation uses.

At-least-once checkpointing mode does not hold records back the same way. It can be appropriate when a few milliseconds matter more than a single aligned snapshot and the sink tolerates duplicates. It is a worse default for a window count. Leave the mode at exactly-once unless you have measured the alignment delay and accepted duplicates.

Under backpressure, alignment can stall: one slow input refuses to deliver its barrier, and the others wait. Unaligned checkpoints let the barrier overtake buffered data and include that buffer in the snapshot. The configuration page states they require exactly-once mode and a maximum of one concurrent checkpoint. They shorten checkpoint time when the graph is stuck. They make snapshots larger because channel state is part of them. Enable them when alignment time is the incident, not as a default on a healthy job.

Checkpoint storage defaults to the JobManager's heap, which is only a demo. Production sets execution.checkpointing.storage to filesystem and execution.checkpointing.dir to a path every TaskManager and the JobManager can write. execution.checkpointing.timeout (default ten minutes on the configuration table) aborts a checkpoint that does not finish. execution.checkpointing.min-pause keeps the job from checkpointing continuously when one snapshot runs long. execution.checkpointing.tolerable-failed-checkpoints defaults to 0, so the first qualifying checkpoint failure fails the job over. Raise it only with an alert, not to hide a full disk.

Externalized checkpoints keep metadata so you can restart after a failure. RETAIN_ON_CANCELLATION also keeps them when an operator cancels the job, and then a human deletes them. DELETE_ON_CANCELLATION removes them on cancel and keeps them if the job fails. Those are the documented retention modes. A cancel that deletes the only snapshot is a surprise during a deploy. Retain on cancellation for production, and own the cleanup.

Savepoints

A savepoint is a snapshot you take on purpose, for a restart with a changed topology or a move between clusters. A checkpoint is the periodic one the coordinator takes for crash recovery. Both can live on a filesystem. They are not interchangeable in operations: you do not delete the checkpoint directory because a savepoint exists unless you have decided the savepoint is the new restore point and you have tested the restore. Flink's operations documentation describes savepoints as the tool for upgrades. Use a savepoint for a planned change of the window definition. Use checkpoints for a TaskManager dying at 02:00.

State

Keyed state is partitioned by key and owned by one parallel instance of one operator. The types the state documentation defines are value, list, map, reducing, and aggregating state. A counter of events per course per window is value state or, more often, the state inside a window operator you do not manage by hand. A session that stores the last N lesson ids per learner is list or map state, and it is the reason people move off a micro-batch that recomputes from scratch every trigger.

Two state backends matter in current Flink. The hashmap backend keeps state on the heap and snapshots it into the checkpoint. RocksDB (EmbeddedRocksDBStateBackend, configuration state.backend.type: rocksdb) keeps keyed state on local disk, with a memory cache, and supports incremental checkpoints (execution.checkpointing.incremental). Incremental checkpoints write a diff. The UI's checkpoint size is then the delta, which the configuration table warns you about, so you do not congratulate yourself for a "small" checkpoint that still depends on a chain of previous files. Use the hashmap backend when state is small and a full snapshot is cheap. Use RocksDB when per-learner session state is larger than you want on the heap. A five-minute count of a few thousand courses does not need RocksDB. The session store might.

Operator state, as opposed to keyed state, belongs to a parallel source task. The Kafka source records partition offsets there. That is the rewind position. It is stored in the checkpoint, not derived from the consumer group at restore time.

Watermarks

A watermark is Flink's statement that event time should not be expected to go backward past that mark, modulo the lateness you built in. Flink SQL expresses it on the table:

textCopy
WATERMARK FOR event_time AS event_time - INTERVAL '10' MINUTE

The subtraction is the allowed lateness. A tumbling window of five minutes closes when the watermark passes the window end. Late rows after that are dropped from the window unless you define a side output, which is a DataStream feature you should only add from the version's documentation. Idle partitions do not advance the watermark. The Kafka page tells you to lower parallelism or to add an idle timeout. A custom watermark that reads event_time out of a JSON payload in PyFlink is the gap the connector page calls out: custom WatermarkStrategy is not supported in PyFlink there. Flink SQL's watermark clause is supported, and it is the path this lesson uses for the tile. If you need a Java DataStream job, WatermarkStrategy.forBoundedOutOfOrderness is the documented Java helper. This course does not pretend the Python binding has that custom strategy.

The Kafka connector, without folklore

The Kafka source builder takes bootstrap servers, topics, a group id, a starting offset initializer, and a deserializer. On each completed checkpoint it can commit offsets to the broker. The page says the source does not use those committed offsets for fault tolerance. They exist so a monitor can see the group. Restore uses the offsets in the checkpoint. If checkpointing is disabled, the source falls back to Kafka's periodic auto-commit, and you have at-most-once or undefined behavior relative to your sink. Turn checkpointing on before you argue about delivery.

The sink's delivery guarantees, from the same page:

EXACTLY_ONCE requires set_transactional_id_prefix (the Java name is setTransactionalIdPrefix). The prefix must be unique among jobs on that cluster so transactions do not collide. Two deployments of the same tile that share a prefix will interfere. Checkpointing must be enabled for both AT_LEAST_ONCE and EXACTLY_ONCE. The default if you forget the guarantee is NONE.

Exactly-once here means one committed effect per record in Kafka for consumers that honor transactions. It does not repair a downstream database that applied the record outside the transaction. It also delays visibility until the checkpoint commits. A one-minute checkpoint interval means a transactional reader can be a minute behind. Shorten the interval only after the checkpoint duration is comfortably smaller than the interval, and set a minimum pause so the job still processes.

When Flink is the wrong proposal

Spark Structured Streaming already computes this course's five-minute tile, with a file sink, a watermark, and a team that knows the Spark UI. Replacing it because Flink is "more streaming" adds a second runtime, a second checkpoint store, and a second on-call surface. Keep Spark when the latency budget is tens of seconds or a minute, the sink is files or an idempotent batch write, state is a windowed counter, and the platform you have is Spark. Propose Flink when you need several of the following together: event-time processing as the native execution model, large keyed state with incremental checkpoints, alignment of sink commits with those checkpoints through a connector that supports it, and latencies that micro-batches are spending on scheduling. A single one of those can still be a Spark job. The combination is the case.

Flink's batch execution exists and is not this lesson. The historical record remains the Spark course_day table from lesson 31 unless a later platform decision moves it. Running two engines does not mean two grains.

Architecture and Design Diagrams

mermaidCopy
flowchart TB
    subgraph jm [JobManager]
        coord[Checkpoint coordinator]
    end

    subgraph tm [TaskManagers]
        src[Kafka source tasks]
        win[Keyed window operator]
        sink[Kafka sink tasks]
        src --> win --> sink
    end

    subgraph durable [Durable storage]
        chk[(Checkpoint directory)]
    end

    topic[(learning.progress)] --> src
    sink --> out[(course.activity.tile topic)]
    coord --> src
    coord --> win
    coord --> sink
    src --> chk
    win --> chk
    sink --> chk

Barriers are the coordinator's instrument. The directory is the restore point. The output topic shows records when the guarantee you picked says to show them.

mermaidCopy
sequenceDiagram
    participant C as Checkpoint coordinator
    participant S as Source task
    participant W as Window task
    participant K as Kafka sink
    C->>S: Barrier N
    S->>W: Barrier N after records in this snapshot
    W->>W: Snapshot keyed window state
    W->>K: Barrier N
    K->>K: Pre-commit transaction when guarantee is exactly-once
    S-->>C: Offsets and state acknowledgements
    C->>K: Commit checkpoint N
    Note over K: Transactional readers see the records now

If the JobManager declares the checkpoint failed, the transaction aborts and the restore replays from checkpoint N-1. Consumers configured with read_committed do not apply the aborted write. Consumers configured with the default read-uncommitted can.

Technical Explanation

A configuration set you can defend

For the live tile on Flink, a starting point that matches the documentation:

Unaligned checkpoints stay off until the UI shows alignment time as the problem. Incremental checkpoints stay off until RocksDB is on.

Windows without a hand-rolled evictor

Flink SQL tumbling windows group rows whose event time falls in a fixed bucket. The watermark on the source table decides when the bucket is complete. A five-minute tumble with a ten-minute watermark has the same product meaning as the Spark tile: a window can emit about ten minutes after event time advances past its end. The numbers should match if both engines ever run during a migration, or the tiles will disagree and both will look broken.

Do not maintain the window by storing every event in a list state and evicting in a process function unless the window definition is not a tumble. The built-in window is the one whose state size you can explain. A process function is justified for the session case: gap duration per learner, last activity timestamp in value state, and a timer. Timers are part of keyed state and are checkpointed with it. A timer you set and forget, with no watermark progress, fires late or not at all. The same idle-source problem as Spark appears here, with a different UI.

Hot keys

course_id skew does not vanish because the engine changed. One key is handled by one subtask. A viral course piles records and state on that subtask. The remedies are the same ideas as lesson 30, expressed as key design: pre-aggregate before a finer key, or isolate the hot course with an explicit split you document. Salting a window key changes the grain unless you merge afterward. Flink's rescale and rebalance calls redistribute records; they do not split a key's state across tasks. State is keyed. Say that in the review so nobody "fixes" skew with a rebalance that the next keyed operator undoes.

Delivery is a chain

Flink checkpoint exactly-once, plus a Kafka sink in EXACTLY_ONCE, plus a consumer on read_committed, is one chain. Break any link and the claim shrinks. A Spark job reading that output topic with failOnDataLoss and its own checkpoint is another hop, with its own cursor. Draw the hops. The word "exactly-once" applies to a hop whose sink and whose reader you named, not to the company.

What this course will not call

There is no TechLearningPro Flink library in these examples. There is no private metric API. Job metrics that matter are the ones Flink and the Kafka connector already expose: checkpoint duration, alignment time, last completed checkpoint, currentEmitEventTimeLag, watermarkLag, and the source idle time. The connector page lists those source metrics. Use them.

Practical Implementation or Design Walkthrough

Problem

Stand up the same five-minute course tile on Flink, with event time taken from the JSON, a ten-minute watermark, and a Kafka sink whose delivery guarantee is a conscious choice. Keep the Spark tile running until this one is measured. Do not decommission Spark in the same change.

SQL table and query

Confirm the connector options against the Flink SQL Kafka connector page for your version. The keys below are the documented ones: connector, topic, properties.bootstrap.servers, properties.group.id, scan.startup.mode, and format.

sqlCopy
-- Flink SQL. Event time is a column, so the watermark does not need a custom
-- PyFlink WatermarkStrategy (unsupported on the DataStream Kafka page).
CREATE TABLE progress (
  event_id STRING,
  course_id STRING,
  lesson_id STRING,
  event_time TIMESTAMP(3),
  WATERMARK FOR event_time AS event_time - INTERVAL '10' MINUTE
) WITH (
  'connector' = 'kafka',
  'topic' = 'learning.progress',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'tlp-flink-course-tile',
  'scan.startup.mode' = 'group-offsets',
  'format' = 'json',
  'json.timestamp-format.standard' = 'ISO-8601'
);

CREATE TABLE course_activity_tile (
  course_id STRING,
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3),
  event_count BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = 'course.activity.tile',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'json'
);

INSERT INTO course_activity_tile
SELECT
  course_id,
  window_start,
  window_end,
  COUNT(*) AS event_count
FROM TABLE(
  TUMBLE(TABLE progress, DESCRIPTOR(event_time), INTERVAL '5' MINUTES)
)
GROUP BY course_id, window_start, window_end;

scan.startup.mode of group-offsets applies when the job has no checkpointed offset. After a checkpoint exists, restore uses the snapshot. The group id is how operators see lag. It is not the cursor. The sink table as written does not set a transactional delivery guarantee; Flink SQL sink options for exactly-once Kafka delivery must be taken from the connector page (sink.delivery-guarantee and a transactional id prefix) before you claim that guarantee. If those options are not set, do not describe this INSERT as exactly-once. The DataStream sink below is where the guarantee is explicit.

PyFlink checkpoint and Kafka sink

pythonCopy
"""Checkpointed Flink job skeleton for the course tile.

Imports and builders follow the Flink checkpointing page and the Kafka connector page.
The window itself is the SQL above when event time lives in the JSON payload.
"""

from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.checkpointing_mode import CheckpointingMode
from pyflink.datastream.connectors.kafka import (
    DeliveryGuarantee,
    KafkaOffsetsInitializer,
    KafkaRecordSerializationSchema,
    KafkaSink,
    KafkaSource,
)
from pyflink.datastream.externalized_checkpoint_retention import (
    ExternalizedCheckpointRetention,
)


def build_env() -> StreamExecutionEnvironment:
    """Methods from the Flink checkpointing page's Python example.

    The directory itself is set with execution.checkpointing.storage and
    execution.checkpointing.dir in Flink configuration, not in this function.
    Leaving those unset keeps snapshots on the JobManager heap.
    """
    env = StreamExecutionEnvironment.get_execution_environment()
    # Interval in milliseconds. Exactly-once is the default mode; set it explicitly.
    env.enable_checkpointing(30_000)
    env.get_checkpoint_config().set_checkpointing_mode(CheckpointingMode.EXACTLY_ONCE)
    env.get_checkpoint_config().set_min_pause_between_checkpoints(10_000)
    env.get_checkpoint_config().set_checkpoint_timeout(120_000)
    env.get_checkpoint_config().set_max_concurrent_checkpoints(1)
    env.get_checkpoint_config().set_tolerable_checkpoint_failure_number(0)
    env.get_checkpoint_config().set_externalized_checkpoint_retention(
        ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION
    )
    return env


def kafka_source(brokers: str) -> KafkaSource:
    return (
        KafkaSource.builder()
        .set_bootstrap_servers(brokers)
        .set_topics("learning.progress")
        .set_group_id("tlp-flink-course-tile")
        .set_starting_offsets(KafkaOffsetsInitializer.committed_offsets())
        .set_value_only_deserializer(SimpleStringSchema())
        .build()
    )


def kafka_sink_exactly_once(brokers: str) -> KafkaSink:
    # Prefix must be unique per job on this Kafka cluster.
    return (
        KafkaSink.builder()
        .set_bootstrap_servers(brokers)
        .set_record_serializer(
            KafkaRecordSerializationSchema.builder()
            .set_topic("course.activity.tile")
            .set_value_serialization_schema(SimpleStringSchema())
            .build()
        )
        .set_delivery_guarantee(DeliveryGuarantee.EXACTLY_ONCE)
        .set_transactional_id_prefix("tlp-course-tile-v1")
        .build()
    )

build_env sets the interval and the retention mode. It does not choose a directory. Set the storage keys in Flink configuration (config.yaml or flink-conf.yaml, depending on the Flink generation you run):

textCopy
execution.checkpointing.storage: filesystem
execution.checkpointing.dir: s3://tlp-checkpoints/prod/flink/course-tile
state.backend.type: hashmap

A job that calls enable_checkpointing and leaves storage on the JobManager heap will lose snapshots when that process exits. The directory is part of the design, not a later cleanup task.

SimpleStringSchema reads and writes strings. It does not parse event time. That is why the watermarked query is the SQL job, and this DataStream skeleton is the checkpoint and delivery demonstration. Wiring env.from_source without a watermark strategy would use the Kafka record timestamp, which is broker append time unless producers set timestamps. Broker append time is not event_time. Do not point the classroom tile at broker time and call it the same metric as the Spark tile.

What to measure before you cut over

Run both tiles against a replay of one hour. Compare counts per (course_id, window_start) after the watermark has closed. Differences should be empty, or explained by events that one job had not yet closed. Record checkpoint duration and alignment time on Flink, and batch duration on Spark. Record how long a transactional consumer waits before a window becomes visible. If that delay is unacceptable, you do not have a requirement for EXACTLY_ONCE visibility; you have a requirement for low latency, and AT_LEAST_ONCE plus an idempotent consumer may fit better. Write that sentence in the decision record.

Enterprise Real-World Scenarios

Scenario 1: The classroom tile that outgrew micro-batches

A live class view needs per-learner session state: the current lesson, the last event time, and whether the learner is still inside a ten-minute gap. The Spark query that tried to hold that state used flatMapGroupsWithState and a RocksDB state provider. It worked, and the batch duration crept from two seconds to twenty as sessions accumulated, because every trigger planned a job. The product's budget was two seconds end to end.

Flink with the hashmap backend, later RocksDB, keyed by learner_key, and a ten-minute session gap implemented with timers, keeps the state on the task and checkpoints it every thirty seconds. The Kafka sink uses AT_LEAST_ONCE because the classroom UI deduplicates on (learner_key, session_start) and the team does not want windows to sit inside an open transaction. The daily completion metric does not read this topic. It still reads the batch fact. The decision record lists the latency budget, the state per key, and the sink guarantee. It also says the Spark tile for the five-minute course count stays, because that one is fine.

Scenario 2: Exactly-once that consumers never observed

A pipeline enabled DeliveryGuarantee.EXACTLY_ONCE and a transactional prefix. Checkpoint duration was healthy. The consumer service used Kafka's default isolation, which reads uncommitted records. During a failed checkpoint the transaction aborted, and the consumer had already applied the records. The team filed the bug against Flink.

The connector's guarantee is for readers that use isolation.level=read_committed. The repair is on the consumer, plus a prefix unique to the job, plus an alert when transactions abort. The Flink job was doing what its setting says. The hop was not exactly-once until the reader joined the contract. This is the same shape as Spark's file-sink claim: the sink's promise has a reader that must match it.

Scenario 3: A second engine without a second grain

A platform proposal replaces every Spark job with Flink so the company has "one streaming standard." The morning engagement join is a sixty-million-row sort-merge against Parquet. Flink can run it. The team that owns it cannot yet operate checkpoints, and the job does not need them. The proposal's cost is two sets of expertise on a batch problem.

The boundary this course uses: Spark for the landed Parquet fact and the minute-scale tile; Flink for the keyed session job whose latency and state say so. A review gate asks which of the four needs — sub-second latency, large keyed state, transactional sink alignment, event-time as the native loop — are actually in the requirement. One checked box is a discussion. Three checked boxes are a Flink job. Zero checked boxes are a Spark job someone tried to make interesting.

Best Practices

• Turn checkpointing on before you discuss delivery. Set a filesystem directory, a timeout, a minimum pause, and retain externalized checkpoints on cancellation, with an owner for cleanup.
• Keep exactly-once checkpoint mode unless alignment latency is a measured problem. Turn on unaligned checkpoints only then, with a single concurrent checkpoint.
• Treat Kafka group offsets as a monitor. Restore from the checkpoint. Test that restore.
• Set the Kafka sink guarantee explicitly. NONE is the default. EXACTLY_ONCE needs a unique transactional prefix and readers on read_committed.
• Define event-time watermarks in Flink SQL when the timestamp is inside the payload and you are on PyFlink. Do not substitute the Kafka record timestamp.
• Match the Spark tile's window and watermark during any overlap, and reconcile closed windows before you turn one of them off.
• Choose hashmap state until the state size says RocksDB. Read incremental checkpoint sizes as deltas.
• Use a savepoint for a planned upgrade. Use checkpoints for crashes. Test both restores on a non-production topic.
• Leave hot-key design in the review. Rebalance does not split keyed state.
• Keep the batch course_day table as the historical grain unless a separate decision moves it.

Common Mistakes and Anti-Patterns

Checkpoint storage left on the JobManager heap. The job survives a TaskManager failure only until the JobManager restarts. Set filesystem storage and a shared directory.

Calling the group offset the cursor. The connector commits offsets for visibility. Recovery reads the checkpoint. A reset of the group without a restore plan creates a second truth.

EXACTLY_ONCE without read_committed consumers, or without a unique transactional prefix. The first leaks aborted transactions. The second collides with another job's transactions. Set both, or pick AT_LEAST_ONCE and make the consumer idempotent.

A custom PyFlink watermark you hoped exists. The Kafka connector page says custom watermark strategies are not supported in PyFlink. Use SQL watermarks or a Java strategy the page documents. Broker time is a different metric.

RocksDB because it sounds durable. Durability is the checkpoint on the filesystem. RocksDB is the local working set. Incremental checkpoints then depend on a chain of files. Use it when heap state is the constraint.

Unaligned checkpoints on a job that is not backpressured. Snapshots get larger and more complex. Wait for alignment time to show up in the metrics.

Deleting checkpoints after a cancel because retention was RETAIN_ON_CANCELLATION and nobody owned cleanup. Or the opposite: DELETE_ON_CANCELLATION during a deploy, and no savepoint. Pick retention, and take a savepoint before a risky upgrade.

Two tiles, two windows, one dashboard. Five minutes in Spark and one minute in Flink, both called "activity," will page you forever. One definition, or two names.

Moving the morning batch join to Flink to unify runtimes. You pay a migration for a workload Spark's planner already fits. Unify where the runtime model matches the requirement.

Salting or rebalancing and expecting keyed state to spread. The key still schedules to one subtask. Split the product key only if the output grain can be merged.

Troubleshooting and Design Review

Symptom: Checkpoints time out. Records stop moving. Alignment time grows.

Likely cause: backpressure on one input, so aligned barriers wait. Or the checkpoint directory is slow.

Investigation: alignment duration versus sync and async checkpoint duration in the UI. Disk and error logs on the checkpoint path. Kafka consumer lag per source subtask.

Corrective action: fix the slow sink or the hot key before you mask it. If alignment itself is the stall and the mode is exactly-once with one concurrent checkpoint, consider unaligned checkpoints. Raise the timeout only after a snapshot has proven it can finish.

Symptom: After a TaskManager loss, counts jumped or dropped.

Likely cause: checkpointing was off, the sink guarantee was NONE, or a consumer read uncommitted records from an aborted transaction. Or the restore did not happen and the job started from latest or earliest group offsets.

Investigation: last completed checkpoint, whether the job actually restored, sink guarantee, consumer isolation.level.

Corrective action: restore from the checkpoint you intended. Align the consumer with the guarantee. Do not "fix" a duplicate by resetting offsets until the sink idempotence story is written down.

Symptom: The Flink tile and the Spark tile disagree on a closed window.

Likely cause: different watermark delays, different time columns (event time versus broker time), or one job still has the window open.

Investigation: print watermark and window end on both. Compare one event_id that landed in one result.

Corrective action: make the definitions match, or stop calling them the same tile. Reprocess the hour from the topic once both watermarks agree.

Symptom: JobManager restarts and the job has no snapshot.

Likely cause: checkpoint storage was the JobManager heap, or the filesystem path was local to one host.

Investigation: execution.checkpointing.storage and the directory listing from another host.

Corrective action: filesystem storage on a shared path. Take a savepoint before the next configuration change so you have a known restore.

Symptom: State backend metrics show a small checkpoint, and a restore fails looking for older files.

Likely cause: incremental checkpoints. The size is a delta. Retention deleted a parent file.

Investigation: execution.checkpointing.incremental and the retained checkpoint history.

Corrective action: retain the chain the backend requires. Read the size as a delta. Do not apply a "delete files older than an hour" policy blindly to that directory.

A design review asks: Why is this not the Spark tile? Where is the checkpoint directory, and what is the restore drill? What is the sink guarantee, the transactional prefix, and the consumer isolation level? What timestamp feeds the watermark? What is the state backend, and is the checkpoint size a full snapshot or a delta? What happens to a hot course_id? Which table remains the system of record overnight? A review that adds Flink to the architecture diagram without those answers has added a runtime.

Interview Questions and Answers

Q1. What do the JobManager and the TaskManagers each own?

The JobManager coordinates the job: scheduling tasks and running the checkpoint coordinator. TaskManagers execute operator tasks, hold keyed state, and snapshot that state when a barrier arrives. Sources and sinks run as tasks too. A checkpoint directory has to be reachable from both, because coordination and bytes are in different processes. High availability for the JobManager is an operational requirement once a job must survive that process exiting. I do not describe the JobManager as the place state lives.

Q2. What is an aligned checkpoint barrier, in operational language?

The coordinator starts checkpoint N. Each source emits a barrier after the records that belong to N. A task waits until it has seen barrier N on every input, snapshots its state, and forwards the barrier. Records that follow the barrier belong to N+1. The snapshot and the source offsets are one restore point. If one input is slow, the others align and wait, which is the backpressure interaction. Unaligned checkpoints let the barrier pass buffered data and record that buffer in the snapshot, under the documented constraints of exactly-once mode and a single concurrent checkpoint.

Q3. Why is exactly-once checkpoint mode not the same sentence as end-to-end exactly-once?

Checkpoint mode says operator state and source positions restore as one snapshot. The sink still has to participate. A Kafka sink in AT_LEAST_ONCE can write a record, crash, restore, and write it again. EXACTLY_ONCE on that sink commits a Kafka transaction when the checkpoint completes, and only consumers reading committed records stay free of the aborted attempt. A database sink outside that transaction needs its own idempotent write. I name the sink and the reader when I say exactly-once. I do not point at CheckpointingMode.EXACTLY_ONCE alone.

Q4. What does the Flink Kafka connector store for recovery, and what is the group commit for?

Recovery offsets live in the checkpoint, with the rest of the snapshot. The source can commit those offsets to Kafka when a checkpoint completes so a consumer group shows progress. The connector documentation says fault tolerance does not rely on that commit. If checkpointing is off, the client falls back toward Kafka auto-commit, and I would not claim a defined relationship to the sink. I monitor the group. I restore the job from the checkpoint directory.

Q5. What has to be true before DeliveryGuarantee.EXACTLY_ONCE is a real setting?

Checkpointing is enabled. The sink is built with that guarantee and with set_transactional_id_prefix set to a prefix no other job uses on the cluster. Consumers that must not see duplicates use isolation.level=read_committed. I accept that those consumers see records only after the checkpoint commits, so the checkpoint interval is part of the latency. If I cannot get a unique prefix or I cannot change the consumers, I use AT_LEAST_ONCE and an idempotent apply. I do not leave the default NONE in place and describe it as at-least-once.

Q6. How do you define the ten-minute watermark in PyFlink without inventing a payload parser the connector lacks?

I use Flink SQL: WATERMARK FOR event_time AS event_time - INTERVAL '10' MINUTE on a table whose event_time column is parsed by the JSON format. The DataStream Kafka page says a custom WatermarkStrategy is not supported in PyFlink. The record timestamp is a poor substitute, because it tracks broker or producer timestamp settings, not the device's event time. A Java job can use WatermarkStrategy.forBoundedOutOfOrderness. I keep the SQL watermark next to the Spark tile's withWatermark("event_time", "10 minutes") so the two definitions can be compared.

Q7. When do you pick RocksDB and incremental checkpoints?

When keyed state, such as per-learner sessions, is too large or too numerous for the hashmap backend's heap snapshot. RocksDB keeps the working set locally and can write incremental checkpoints, which are diffs. The configuration table warns that the reported size is the delta. I retain the file chain the restore needs. I do not turn RocksDB on for a few thousand course counters. The hashmap backend and a full snapshot are simpler there. Durability in both cases is the checkpoint on the shared filesystem.

Q8. How does a savepoint differ from a checkpoint in an upgrade?

Checkpoints are the periodic crash-recovery snapshots. Savepoints are the ones you trigger for a planned change: a new window, a new parallelism, a move. I take a savepoint, stop the job, and restore from that savepoint on the new version after I have tested it. I do not delete the last checkpoint until that restore is proven, and I do not treat a random checkpoint directory as a savepoint just because both are files. Externalized checkpoint retention still needs an owner so RETAIN_ON_CANCELLATION does not fill the bucket, and so DELETE_ON_CANCELLATION does not surprise a deploy.

Q9. A hot course id pins one TaskManager's CPU. Will a rebalance fix it?

It redistributes records until the next keyed operator. Keyed window state for that course_id still lives on one subtask. I either accept the skew, isolate that course, or change the key in a way the output can merge. Adding TaskManagers without changing the key adds idle workers. This is the same fact as Spark's hash partitioner: one key, one partition. The UI will show one busy subtask and a calm watermark on the others, which can also stall the watermark if idleness is not set.

Q10. When do you keep Spark Structured Streaming?

When the latency budget fits a micro-batch, the sink is the file sink or another idempotent batch write, and the state is a windowed aggregate the Spark checkpoint already handles. The five-minute course tile in lesson 32 is that job. I add Flink when the requirement stacks large keyed state, transactional sink alignment, and a latency the micro-batch spends on scheduling. I do not migrate the morning Parquet join onto Flink to reduce the number of logos on a slide. Two engines are justified by two workload shapes, and the historical grain stays a single table.

Q11. What do you reconcile if both tiles run for a week?

Closed windows only, on (course_id, window_start), after both watermarks have passed the window end. I expect late events and still-open windows to explain small gaps, and I expect zero unexplained gaps once timestamps match. I do not reconcile the Flink topic against an open window. I do not sum either tile to replace course_day. The batch table is still the system of record for the date. The dual-run ends when the decision record says which live tile remains, not when the new one has merely started.

Hands-On Exercises

Exercise: A decision record and a checkpointed skeleton

Problem statement. You do not have to run a cluster to complete the reasoning. You do have to produce a Flink configuration and a side-by-side definition with the Spark tile. If you have a local Flink mini-cluster, restore from a checkpoint once.

Tasks.

1. Copy the Spark tile's window, watermark delay, and grain next to the Flink SQL tumble. Mark any difference as a defect.
2. Set checkpoint interval, mode, minimum pause, timeout, filesystem directory, and RETAIN_ON_CANCELLATION using the documented methods or configuration keys. State what the JobManager heap default would lose.
3. Choose a Kafka sink guarantee. If you choose EXACTLY_ONCE, name the transactional prefix and the consumer isolation level. If you choose AT_LEAST_ONCE, name the idempotent key the consumer uses.
4. Explain where recovery offsets live, and why resetting the consumer group is not a restore.
5. Write the four-point test (latency, keyed state, transactional sink, native event-time loop) for the classroom session job and for the existing five-minute tile. Recommend an engine for each.
6. Describe one hot-key failure and a response that does not pretend rebalance splits state.
7. If you can run Flink locally, cancel and restore. If you cannot, write the restore steps you would execute and the metric that tells you the restore worked.

Expected deliverables. The SQL, the configuration, the guarantee paragraph, and the two-engine recommendation.

Evaluation rubric.

You pass when a reviewer can point to the checkpoint directory, the delivery chain, and a reason the morning Spark job stays where it is.

Lesson Summary

Flink executes a continuous operator graph. The JobManager coordinates checkpoints; TaskManagers hold state. An aligned barrier produces a snapshot of state and source positions. That is exactly-once state recovery when the source can rewind. End-to-end exactly-once adds a sink that commits with the checkpoint and a reader that honors the commit. On Kafka, that is DeliveryGuarantee.EXACTLY_ONCE, a unique transactional prefix, and read_committed consumers. Group offsets are a monitor. Watermarks for a JSON event time belong in Flink SQL when you are in PyFlink. RocksDB and unaligned checkpoints solve measured problems, state size and alignment delay, and they add file chains and snapshot complexity.

The five-minute course tile can remain on Spark. A per-learner session with a two-second budget is the shape that justifies a second engine. The next lesson, Distributed Processing Reliability, pulls Spark's lineage, Spark's streaming checkpoint, and Flink's barriers into one failure model: what a retry may repeat, what a sink must ignore, and which domain failed.

Further Reading

• Flink checkpointing — enable_checkpointing, modes, timeouts, externalized retention, and unaligned checkpoints.
• Flink Kafka connector — source offsets, PyFlink builders, and DeliveryGuarantee.
• Flink state backends — hashmap and RocksDB, and what incremental checkpoint sizes mean.
• Flink SQL Kafka connector — the WITH options used on the progress table. Confirm keys for your version.
• Structured Streaming with Spark — the tile this lesson is allowed to replace only after a reconciliation.
• PyFlink CheckpointingMode — the enum the configuration calls.