Apache Flink Tutorial — Stream Processing, Setup & Usage

Core Concepts

The building blocks of stateful stream processing.

Streams

Unbounded streams are continuous, real-time data flows (e.g., click events). Bounded streams are finite datasets (e.g., batch files). Flink handles both with the same API.

Stateful Processing

Operators can maintain local state (counters, windows, custom objects) across events. State is fault-tolerant via checkpointing and can be queried externally.

Event Time & Watermarks

Event time: When an event actually occurred (embedded timestamp). Watermarks: Special markers that track progress in event time, allowing Flink to handle late-arriving and out-of-order events.

Checkpointing

Flink periodically snapshots the state of all operators to durable storage. On failure, it restores from the last checkpoint, providing exactly-once processing guarantees.

Windowing

Group events into finite buckets for aggregation: Tumbling (fixed, non-overlapping), Sliding (overlapping), Session (activity-based gaps), Global (custom triggers).

DataStream API

The primary Java/Scala/Python API for building streaming applications. Supports declarative transformations: map, filter, keyBy, window, reduce, process.

Savepoints

User-triggered, portable snapshots of application state. Unlike checkpoints (automatic), savepoints enable planned upgrades, schema migrations, and A/B testing without data loss.

Connectors

Built-in source/sink connectors for Kafka, Kinesis, JDBC, Elasticsearch, S3/HDFS, and more. Custom connectors via the Source/Sink API.

Architecture

The master-worker model that powers Flink's distributed execution.

🧠

JobManager

The master process. Receives job submissions, builds the execution graph, coordinates checkpoints, manages recovery on failures, and schedules tasks across TaskManagers.

⚙️

TaskManagers

The workers. Each has a fixed number of task slots that execute operator subtasks in parallel. They manage local state backends and exchange data via network buffers.

💾

State Backends

HashMapStateBackend: In-memory, fastest for small state. EmbeddedRocksDBStateBackend: Disk-backed, supports state larger than available memory. Checkpoints to S3/HDFS/GCS.

                    /* Execution Flow */

                    Client ──submit──→ JobManager ──schedule──→ TaskManager 1 [Slot 0] [Slot 1]

                                                        ──schedule──→ TaskManager 2 [Slot 0] [Slot 1]

                    JobManager ──checkpoint barrier──→ Source → Map → KeyBy → Window → Sink

                                                                 ↓ snapshot state ↓

                                                            S3 / HDFS / GCS (checkpoint storage)

Setup Guide

Start a Flink cluster locally or via Docker.

Option 1: Docker (Recommended)

                        # docker-compose.yml

                        version: '3'

                        services:

                          jobmanager:

                            image: flink:1.20-java17

                            ports: ["8081:8081"]

                            command: jobmanager

                            environment:

                              FLINK_PROPERTIES: |

                                jobmanager.rpc.address: jobmanager

                          taskmanager:

                            image: flink:1.20-java17

                            depends_on: [jobmanager]

                            command: taskmanager

                            environment:

                              FLINK_PROPERTIES: |

                                jobmanager.rpc.address: jobmanager

                                taskmanager.numberOfTaskSlots: 4

                        # Start cluster

                        docker compose up -d

                        # Dashboard at http://localhost:8081

Option 2: Native Install

                        # Prerequisites: Java 17+

                        java -version

                        # Download Flink

                        wget https://dlcdn.apache.org/flink/

                          flink-1.20.1/flink-1.20.1-bin-scala_2.12.tgz

                        tar -xzf flink-*.tgz

                        cd flink-1.20.1

                        # Start local cluster

                        ./bin/start-cluster.sh

                        # Dashboard at http://localhost:8081

                        # Submit a job

                        ./bin/flink run \

                          ./examples/streaming/WordCount.jar

                        # Stop cluster

                        ./bin/stop-cluster.sh

Usage Examples

DataStream API, windowing, and Kafka integration.

Java — DataStream API

                        StreamExecutionEnvironment env =

                          StreamExecutionEnvironment

                            .getExecutionEnvironment();

                        // Enable checkpointing every 60s

                        env.enableCheckpointing(60000);

                        // Read from Kafka source

                        KafkaSource<String> source = KafkaSource

                          .<String>builder()

                          .setBootstrapServers("localhost:9092")

                          .setTopics("user-events")

                          .setGroupId("flink-consumer")

                          .setValueOnlyDeserializer(

                            new SimpleStringSchema())

                          .build();

                        DataStream<String> stream = env

                          .fromSource(source, 

                            WatermarkStrategy.noWatermarks(),

                            "Kafka Source");

                        // Transform: count events per user

                        stream.map(event -> parseUser(event))

                          .keyBy(event -> event.userId)

                          .window(TumblingEventTimeWindows

                            .of(Time.minutes(5)))

                          .reduce((a, b) -> merge(a, b))

                          .print();

                        env.execute("User Event Counter");

Python — PyFlink

                        # pip install apache-flink

                        from pyflink.datastream import \

                          StreamExecutionEnvironment

                        from pyflink.common import WatermarkStrategy

                        from pyflink.common.serialization import \

                          SimpleStringSchema

                        env = StreamExecutionEnvironment \

                          .get_execution_environment()

                        env.set_parallelism(4)

                        # Process a bounded dataset

                        ds = env.from_collection([

                          ("Alice", 1), ("Bob", 2),

                          ("Alice", 3), ("Bob", 1)

                        ])

                        # Key by name, sum values

                        result = ds \

                          .key_by(lambda x: x[0]) \

                          .reduce(lambda a, b: 

                            (a[0], a[1] + b[1]))

                        result.print()

                        env.execute("PyFlink Word Count")

                        # Output:

                        # ("Alice", 4)

                        # ("Bob", 3)

Best Practices

Checkpoint Tuning

Set checkpoint interval based on recovery time objectives. Use incremental checkpoints with RocksDB for large state. Configure execution.checkpointing.timeout to catch stalls.

State Backend Selection

Use HashMapStateBackend for fast access with small state (<1 GB). Switch to RocksDB when state exceeds memory capacity. RocksDB supports incremental checkpoints.

Parallelism Config

Set parallelism equal to the number of Kafka partitions for source operators. Over-parallelizing wastes slots; under-parallelizing creates bottlenecks. Use setParallelism() per operator.

Handle Late Data

Configure allowedLateness() on windows to process late events. Use side outputs to capture and separately process events that arrive after the watermark.

Use Savepoints for Upgrades

Before deploying a new version, take a savepoint with flink savepoint <jobId>. Restart from the savepoint to ensure zero data loss during upgrades. Assign stable UIDs to operators.

Monitor Backpressure

Use the Flink Web UI to detect backpressure (tasks shown in red). Backpressure indicates a slow downstream operator. Fix by increasing parallelism, optimizing the bottleneck, or adjusting buffer sizes.

Scaling & Production

Parallelism Tuning

Set per-operator parallelism based on throughput needs. Source parallelism = Kafka partitions. CPU-bound operators need more slots. I/O-bound sinks can use fewer. Use setParallelism() per operator, not just globally.

Reactive Scaling

Flink's Reactive Mode automatically adjusts job parallelism when TaskManagers are added or removed. Combine with Kubernetes HPA to scale TaskManager pods based on CPU or custom Kafka lag metrics.

State Backend at Scale

Use RocksDB state backend for jobs with large state (GB+). It spills to disk and supports incremental checkpointing — only changed SST files are uploaded, reducing checkpoint time from minutes to seconds at TB-scale state.

Rescaling State

Flink supports state rescaling when changing parallelism. Key groups are redistributed across new subtasks. Take a savepoint before rescaling, then restart with new parallelism. Keyed state and operator state handle redistribution differently.

Network Buffer Tuning

Flink's network stack uses buffer pools for data exchange. Increase taskmanager.network.memory.fraction (default 0.1) for high-throughput jobs. Too few buffers cause backpressure; too many waste memory. Monitor via the Flink Web UI.

Cluster Sizing

Rule of thumb: 1 CPU core per slot, 2–4 GB RAM per slot. A job with parallelism 32 needs 32 slots across TaskManagers. Allocate extra capacity for checkpoint overhead, GC pauses, and burst traffic (1.5× baseline).

Frequently Asked Questions

Flink vs Spark Streaming — when to use which?→

Use Flink for true event-at-a-time processing with low latency (milliseconds), complex event processing, and exactly-once guarantees. Use Spark Structured Streaming if you have a heavy batch/ML workload and want a unified batch+stream engine, or if your team already uses Spark extensively. Flink excels at stateful streaming; Spark excels at batch analytics with streaming as an add-on.

Does Flink guarantee exactly-once processing?→

Yes — Flink provides exactly-once state consistency through its checkpointing mechanism. For end-to-end exactly-once (including sinks), both the source (e.g., Kafka) and sink (e.g., Kafka, JDBC) must support transactions. Flink's TwoPhaseCommitSinkFunction implements this pattern.

Can I use Python with Flink?→

Yes — PyFlink provides full access to Flink's DataStream and Table APIs from Python. Install via pip install apache-flink. PyFlink runs Python UDFs using an optimized cross-process communication layer. It's production-ready for data engineering but Java/Scala still offer the best performance for latency-sensitive workloads.

What happened to the DataSet API?→

Flink 2.0 (2025) removed the legacy DataSet API in favor of a unified stream-first approach. Bounded (batch) data is now processed as a special case of streaming using the DataStream API. This simplification means one API for both real-time and historical data processing.

How does Flink handle out-of-order events?→

Flink uses watermarks to declare "all events up to timestamp T have arrived." Events arriving after their watermark are considered late. You can configure allowedLateness() on windows to still process them, and use side outputs to capture events that arrive too late even for the allowed lateness period.