Apache Flink Tutorial: Stateful Stream Processing (2026)
Apache Flink is the gold standard for stateful stream processing, and after building real-time analytics platforms with it for several years, I can confirm that nothing else matches its combination of exactly-once semantics, state management, and event-time processing. Flink processes data as it arrives, maintains state across events, and handles late data gracefully.
This tutorial covers Flink's architecture, DataStream API, state management, windowing, time semantics, and production deployment strategies for building robust streaming applications.
Flink Architecture and Execution Model
Flink runs on a distributed cluster with a JobManager and TaskManagers. The JobManager coordinates job execution, manages checkpoints, and handles failover. TaskManagers execute the actual data processing tasks. Flink programs compile into a dataflow graph of operators, which are distributed across TaskManagers for parallel execution.
Flink's runtime is based on the concept of a 'stream' as the fundamental data abstraction. Batch processing is a special case of stream processing where the input is bounded. This unified model means the same API handles both real-time and historical data.
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);
DataStream stream = env
.socketTextStream("localhost", 9999)
.flatMap(new Tokenizer())
.keyBy(value -> value.f0)
.window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
.sum(1);
stream.print();
env.execute("WordCount Streaming Job"); State Management and Checkpointing
State is the memory of a streaming application. Flink manages state through state backends: HashMapStateBackend (heap-based, fast but limited by memory) and EmbeddedRocksDBStateBackend (disk-based, scalable for large state). Checkpointing periodically snapshots state to durable storage (S3, HDFS) for fault tolerance.
Flink's checkpoint mechanism uses a distributed snapshot algorithm (Chandy-Lamport). When a checkpoint is triggered, operators snapshot their current state and barrier tokens flow through the dataflow graph. If a failure occurs, Flink replays from the last successful checkpoint, restoring exact state.
// State with RocksDB backend
StateBackend stateBackend = new EmbeddedRocksDBStateBackend(true);
env.setStateBackend(stateBackend);
env.getCheckpointConfig().setCheckpointStorage("s3://bucket/checkpoints");
// Keyed state
public class CountFunction extends KeyedProcessFunction {
private transient ValueState countState;
@Override
public void open(Configuration parameters) {
ValueStateDescriptor desc = new ValueStateDescriptor<>("count", Long.class);
countState = getRuntimeContext().getState(desc);
}
@Override
public void processElement(Event event, Context ctx, Collector out) throws Exception {
Long count = countState.value();
if (count == null) count = 0L;
countState.update(count + 1);
if (count + 1 > 100) {
out.collect(new Alert("Threshold exceeded for " + ctx.getCurrentKey()));
}
}
} Windowing and Time Semantics
Windows group events into finite buckets for aggregation. Flink supports tumbling, sliding, session, and global windows. Event-time processing uses watermarks to track progress through the event stream. A watermark with timestamp T means no events with timestamp less than T are expected to arrive.
Late events (arriving after their window has been computed) can be handled using allowed lateness or side outputs. This is critical for real-world systems where network delays and out-of-order delivery are common.
// Tumbling window with event time
DataStream scores = ...;
scores
.keyBy(Score::getPlayerId)
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.allowedLateness(Time.minutes(1))
.sideOutputLateData(lateOutputTag)
.process(new ProcessWindowFunction() {
@Override
public void process(String key, Context context, Iterable elements, Collector out) {
double avg = StreamSupport.stream(elements.spliterator(), false)
.mapToDouble(Score::getValue).average().orElse(0.0);
out.collect(new Result(key, avg, context.window().getEnd()));
}
}); DataStream and Table APIs
The DataStream API provides low-level control over state, timers, and event processing. The Table API provides a higher-level, SQL-like abstraction that Flink optimizes through its query planner. Both APIs can be mixed in the same program — you can convert between DataStream and Table seamlessly.
The Table API is preferred for analytical queries where SQL expressiveness reduces boilerplate. The DataStream API is preferred for complex event processing, custom stateful logic, and integration with external systems.
// Table API example
Table table = tableEnv.fromDataStream(eventStream);
Table result = table
.filter($("event_type").isEqual("purchase"))
.groupBy($("user_id"))
.select($("user_id"), $("amount").sum().as("total"))
.filter($("total").greaterThan(1000));
tableEnv.toDataStream(result).print();
// Flink SQL
// CREATE TABLE orders (
// user_id STRING,
// amount DOUBLE,// event_time TIMESTAMP(3),
// WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
// ) WITH ('connector' = 'kafka', 'topic' = 'orders', ...);
// SELECT user_id, SUM(amount) FROM orders
// WHERE event_time > CURRENT_TIMESTAMP - INTERVAL '1' HOUR
// GROUP BY user_id;Connectors and Integration
Flink's connector ecosystem integrates with virtually every data system. Kafka connectors handle source/sink with exactly-once guarantees. JDBC connectors read/write databases in batch or streaming mode. File connectors support Parquet, ORC, and Avro formats. Elasticsearch, Cassandra, and Redis connectors provide low-latency sinks.
When choosing connectors, consider the delivery guarantee: Kafka provides exactly-once with transactional sinks; JDBC provides at-least-once with idempotent writes; filesystem provides exactly-once with atomic commits.
// Kafka source
KafkaSource source = KafkaSource.builder()
.setBootstrapServers("kafka:9092")
.setTopics("events")
.setGroupId("flink-consumer")
.setStartingOffsets(OffsetsInitializer.latest())
.setValueOnlyDeserializer(new EventDeserializer())
.build();
DataStream stream = env.fromSource(source, WatermarkStrategy
.forBoundedOutOfOrderness(Duration.ofSeconds(10)), "Kafka Source");
// JDBC sink
JdbcSink.sink(
"INSERT INTO users (id, name, email) VALUES (?, ?, ?) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name",
(ps, user) -> {
ps.setString(1, user.getId());
ps.setString(2, user.getName());
ps.setString(3, user.getEmail());
},
JdbcExecutionOptions.builder().withBatchSize(1000).build(),
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:postgresql://db:5432/mydb")
.withDriverName("org.postgresql.Driver")
.build()
); Performance Tuning and Deployment
Flink performance depends on state backend choice, checkpoint interval, network buffer configuration, and task chaining. RocksDB state backend is recommended for state larger than memory. Checkpoint intervals should balance fault tolerance frequency against overhead — too frequent checkpoints waste I/O, too infrequent checkpoints increase recovery time.
Deploy on Kubernetes using the Flink Kubernetes Operator for production. The operator manages JobManager/TaskManager pods, handles upgrades, and provides CRDs for job submission. For session clusters, use standalone or YARN; for application mode, use Kubernetes.
# Flink configuration for production
#
# State backend
state.backend: rocksdb
state.backend.rocksdb.memory.managed: true
state.backend.incremental: true
# Checkpointing
execution.checkpointing.interval: 60000
execution.checkpointing.min-pause: 30000
execution.checkpointing.timeout: 600000
state.checkpoints.num-retained: 3
# Network
taskmanager.network.memory.fraction: 0.1
taskmanager.network.memory.min: 64mb
# Kubernetes deployment
apiVersion: flink.apache.org/v1
kind: FlinkDeployment
spec:
image: my-flink:1.18
flinkVersion: v1_18
serviceAccount: flink
jobManager:
resource:
memory: 2048m
cpu: 1
taskManager:
resource:
memory: 4096m
cpu: 2
job:
jarURI: local:///opt/flink/usrlib/my-job.jar
entryClass: com.example.MyStreamingJobFrequently Asked Questions
What is the difference between Flink and Spark Structured Streaming?
Flink processes events one at a time with native event-time support. Spark processes micro-batches, adding latency. Flink's state management is more mature, and its exactly-once semantics are built into the runtime rather than bolted on.
How does Flink handle late data?
Flink uses watermarks to track event-time progress. Late events can be handled via allowed lateness (extending window retention) or side outputs (routing late events for separate processing).
Can Flink replace Kafka Streams?
Flink is better for complex stateful processing, large state, and multi-source aggregation. Kafka Streams is simpler for applications that only need to read from and write to Kafka with minimal dependencies.
What state backend should I use?
HashMapStateBackend for small state (<10GB). EmbeddedRocksDBStateBackend for large state. RocksDB supports incremental checkpoints, reducing checkpoint time for large state.
Originally published on Ayodhyyya. Last updated June 1, 2026.