Flink - Exactly Once

Official Document

An Aside

Spark 2.3 was recently released, and stream processing is no longer in beta. Spark’s Structured Streaming now has almost all of Flink’s features. A strong ecosystem and community really make a difference. Flink needs to pick up the pace…

Introduction

Exactly-once is a frequently discussed guarantee and an ideal that we try to achieve in software development. In distributed programs, it can mean several things: reading a source exactly once, processing exactly once, or storing data exactly once. The usual interpretation, however, is end-to-end semantics: each input affects the output only once.

Exacly-once

For a distributed program to achieve exactly-once semantics, the output layer should allow only two outcomes: all outputs are committed together, or none are. Flink provides the TwoPhaseCommitSinkFunction abstract class for this purpose: JIRA, PR. As the name suggests, this class uses checkpointing to provide two phases:

1. preCommit()
2. commit()

preCommit() Diagram:

This diagram shows a dataflow that reads from Kafka, processes data with a WindowFunction, and writes it back to Kafka. During a checkpoint, a barrier first divides the stream into two parts at the data source, then propagates downstream to trigger snapshots. This ensures that all snapshots cover the same portion of the data. The final stage calls the sink’s preCommit(), indicating that the sinks on all TaskManagers can prepare for commit().

They cannot commit yet, because the TaskManagers do not know whether the other nodes are ready to commit!

commit() Diagram:

After a checkpoint finishes, the JobManager notifies listeners to invoke their callbacks. When a sink receives the JobManager’s checkpointCompleted notification, it knows that every node is ready to commit. Only then does the actual commit take place.

Usage in Detail

In simple terms, this is the familiar idea of achieving exactly-once through a two-step output process. Hive, for example, first writes files into a temporary directory, then moves them into the target directory.

In code, this becomes four steps:

• beginTransaction - Begin the transaction, for example by creating a temporary directory
• preCommit - Prepare for the final commit, for example by flushing data into the temporary directory
• commit - Commit the data, for example by moving it from the temporary directory into the final directory
• abort - Delete the temporary directory

··· I will write and post an example later. ···