Savepoints

A savepoint is basically identical to a checkpoint—it is a consistent and complete snapshot of an application’s state. However, the lifecycles of checkpoints and savepoints differ. Checkpoints are automatically created, loaded in case of a failure, and automatically removed by Flink (depending on the configuration of the application). Moreover, checkpoints are automatically deleted when an application is canceled, unless the application explicitly enabled checkpoint retention. In contrast, savepoints must be manually triggered by a user or an external service and are never automatically removed by Flink.

A savepoint is a directory in a persistent data storage. It consists of a subdirectory that holds the data files containing the states of all tasks and a binary metadata file that includes absolute paths to all data files. Because the paths in the metadata file are absolute, moving a savepoint to a different path will render it unusable. Here is the structure of a savepoint:

# Savepoint root path
/savepoints/

# Path of a particular savepoint
/savepoints/savepoint-:shortjobid-:savepointid/

# Binary metadata file of a savepoint
/savepoints/savepoint-:shortjobid-:savepointid/_metadata

# Checkpointed operator states
/savepoints/savepoint-:shortjobid-:savepointid/:xxx

Managing Applications with the Command-Line Client

Flink’s command-line client provides the functionality to start, stop, and manage Flink applications. It reads its configuration from the ./conf/flink-conf.yaml file (see “System Configuration”). You can call it from the root directory of a Flink setup with the command./bin/flink.

When run without additional parameters, the client prints a help message.

Starting an application

You can start an application with the run command of the command-line client:

./bin/flink run ~/myApp.jar

The above command starts the application from the main() method of the class that is referenced in the program-class property of the JAR file’s META-INF/MANIFEST.MF file without passing any arguments to the application. The client submits the JAR file to the master process, which distributes it to the worker nodes.

You can pass arguments to the main() method of an application by appending them at the end of the command:

./bin/flink run ~/myApp.jar my-arg1 my-arg2 my-arg3

By default, the client does not return after submitting the application but waits for it to terminate. You can submit an application in detached mode with the -d flag as shown here:

./bin/flink run -d ~/myApp.jar

Instead of waiting for the application to terminate, the client returns and prints the JobID of the submitted job. The JobID is used to specify the job when taking a savepoint, canceling, or rescaling an application. You can specify the default parallelism of an application with the -p flag:

./bin/flink run -p 16 ~/myApp.jar

The above command sets the default parallelism of the execution environment to 16. The default parallelism of an execution environment is overwritten by all settings explicitly specified by the source code of the application—the parallelism that is defined by calling setParallelism() on StreamExecutionEnvironment or on an operator has precedence over the default value.

If the manifest file of your application JAR file does not specify an entry class, you can specify the class using the -c parameter:

./bin/flink run -c my.app.MainClass ~/myApp.jar

The client will try to start the static main() method of the my.app.MainClass class.

By default, the client submits an application to the Flink master specified by the ./conf/flink-conf.yaml file (see the configuration for different setups in “System Configuration”). You can submit an application to a specific master process using the -m flag:

./bin/flink run -m myMasterHost:9876 ~/myApp.jar

This command submits the application to the master that runs on host myMasterHost at port 9876.

./bin/flink list -r
Waiting for response...
------------------ Running/Restarting Jobs -------------------
17.10.2018 21:13:14 : bc0b2ad61ecd4a615d92ce25390f61ad : 
Socket Window WordCount (RUNNING)
​--------------------------------------------------------------

./bin/flink savepoint <jobId> [savepointPath]

./bin/flink savepoint bc0b2ad61ecd4a615d92ce25390f61ad 
hdfs:///xxx:50070/savepoints
Triggering savepoint for job bc0b2ad61ecd4a615d92ce25390f61ad.
Waiting for response...
Savepoint completed. 
Path: hdfs:///xxx:50070/savepoints/savepoint-bc0b2a-63cf5d5ccef8
You can resume your program from this savepoint with the run command.

./bin/flink savepoint -d <savepointPath>

./bin/flink savepoint -d 
hdfs:///xxx:50070/savepoints/savepoint-bc0b2a-63cf5d5ccef8
Disposing savepoint 'hdfs:///xxx:50070/savepoints/savepoint-bc0b2a-63cf5d5ccef8'.
Waiting for response...
​Savepoint 'hdfs:///xxx:50070/savepoints/savepoint-bc0b2a-63cf5d5ccef8' disposed.

Canceling an application

An application can be canceled in two ways: with or without a savepoint. To cancel a running application without taking a savepoint run the following command:

./bin/flink cancel <jobId>

In order to take a savepoint before canceling a running application add the -s flag to the cancel command:

./bin/flink cancel -s [savepointPath] <jobId>

If you do not specify a savepointPath, the default savepoint directory as configured in the ./conf/flink-conf.yaml file is used (see “System Configuration”). The command fails if the savepoint folder is neither explicitly specified in the command nor available from the configuration. To cancel the application with the JobID bc0b2ad61ecd4a615d92ce25390f61ad and store the savepoint at hdfs:///xxx:50070/savepoints, run the command:

./bin/flink cancel -s 
hdfs:///xxx:50070/savepoints d5fdaff43022954f5f02fcd8f25ef855
Cancelling job bc0b2ad61ecd4a615d92ce25390f61ad 
with savepoint to hdfs:///xxx:50070/savepoints.
Cancelled job bc0b2ad61ecd4a615d92ce25390f61ad. 
Savepoint stored in hdfs:///xxx:50070/savepoints/savepoint-bc0b2a-d08de07fbb10.

Starting an application from a savepoint

Starting an application from a savepoint is fairly simple. All you have to do is start an application with the run command and additionally provide a path to a savepoint with the -s option:

./bin/flink run -s <savepointPath> [options] <jobJar> [arguments]

When the job is started, Flink matches the individual state snapshots of the savepoint to all states of the started application. This matching is done in two steps. First, Flink compares the unique operator identifiers of the savepoint and application’s operators. Second, it matches for each operator the state identifiers (see “Savepoints” for details) of the savepoint and the application.

As mentioned, an application can only be started from a savepoint if it is compatible with the savepoint. An unmodified application can always be restarted from its savepoint. However, if the restarted application is not identical to the application from which the savepoint was taken, there are three cases to consider:

• If you added a new state to the application or changed the unique identifier of a stateful operator, Flink will not find a corresponding state snapshot in the savepoint. In this case, the new state is initialized as empty.

• If you removed a state from the application or changed the unique identifier of a stateful operator, there is state in the savepoint that cannot be matched to the application. In this case, Flink does not start the application to avoid losing the state in the savepoint. You can disable this safety check by adding the -n option to the run command.

• If you changed a state in the application—changed the state primitive or modified the data type of the state—the application fails to start. This means that you cannot easily evolve the data type of a state in your application, unless you designed your application with state evolution in mind from the start. The Flink community is currently working on improving the support for state evolution. (See “Modifying the State of an Operator”.)

Scaling an Application In and Out

Decreasing or increasing the parallelism of an application is not difficult. You need to take a savepoint, cancel the application, and restart it with an adjusted parallelism from the savepoint. The state of the application is automatically redistributed to the larger or smaller number of parallel operator tasks. See “Scaling Stateful Operators” for details on how the different types of operator state and keyed state are scaled. However, there are a few things to consider.

If you require exactly-once results, you should take the savepoint and stop the application with the integrated savepoint-and-cancel command. This prevents another checkpoint from completing after the savepoint, which would trigger exactly-once sinks to emit data after the savepoint.

As discussed in “Setting the Parallelism”, the parallelism of an application and its operators can be specified in different ways. By default, operators run with the default parallelism of their associated StreamExecutionEnvironment. The default parallelism can be specified when starting an application (e.g., using the -p parameter in the CLI client). If you implement the application such that the parallelism of its operators depends on the default environment parallelism, you can simply scale an application by starting it from the same JAR file and specifying a new parallelism. However, if you hardcoded the parallelism on the StreamExecutionEnvironment or on some of the operators, you might need to adjust the source code and recompile and repackage your application before submitting it for execution.

If the parallelism of your application depends on the environment’s default parallelism, Flink provides an atomic rescale command that takes a savepoint, cancels the application, and restarts it with a new default parallelism:

./bin/flink modify <jobId> -p <newParallelism>

To rescale the application with the jobId bc0b2ad61ecd4a615d92ce25390f61ad to a parallelism of 16, run the command:

./bin/flink modify bc0b2ad61ecd4a615d92ce25390f61ad -p 16
Modify job bc0b2ad61ecd4a615d92ce25390f61ad.
​Rescaled job bc0b2ad61ecd4a615d92ce25390f61ad. Its new parallelism is 16.

As described in “Scaling Stateful Operators”, Flink distributes keyed state on the granularity of so-called key groups. Consequently, the number of key groups of a stateful operator determines its maximum parallelism. The number of key groups is configured per operator using the setMaxParallelism() method. (See “Defining the Maximum Parallelism of Keyed State Operators”.)

Managing Applications with the REST API

The REST API can be directly accessed by users or scripts and exposes information about the Flink cluster and its applications, including metrics as well as endpoints to submit and control applications. Flink serves the REST API and the Web UI from the same web server, which runs as part of the Dispatcher process. By default, both are exposed on port 8081. You can configure a different port at the ./conf/flink-conf.yaml file with the configuration key rest.port. A value of -1 disables the REST API and Web UI.

A common command-line tool to interact with REST API is curl. A typical curl REST command looks like:

curl -X <HTTP-Method> [-d <parameters>] http://hostname:port/v1/<REST-point>

The v1 indicates the version of the REST API. Flink 1.7 exposes the first version (v1) of the API. Assuming you are running a local Flink setup that exposes its REST API on port 8081, the following curl command submits a GET request to the /overview REST point:

curl -X GET http://localhost:8081/v1/overview

The command returns some basic information about the cluster, such as the Flink version, the number of TaskManagers, slots, and jobs that are running, finished, cancelled, or failed:

{
 "taskmanagers":2,
 "slots-total":8,
 "slots-available":6,
 "jobs-running":1,
 "jobs-finished":2,
 "jobs-cancelled":1,
 "jobs-failed":0,
 "flink-version":"1.7.1",
 "flink-commit":"89eafb4"
}

In the following, we list and briefly describe the most important REST calls. Refer to the official documentation of Apache Flink for a complete list of supported calls. “Managing Applications with the Command-Line Client” provides more details about some of the operations, such as upgrading or scaling an application.

curl -X GET http://hostname:port/v1/jobmanager/metrics?get=metric1,metric2

GET /taskmanagers/<tmId>/metrics

curl -X GET http://hostname:port/v1/taskmanagers/<tmId>/metrics?get=metric1

curl -X POST -H "Expect:" -F "jarfile=@path/to/flink-job.jar" 
 http://hostname:port/v1/jars/upload

curl -d '{"parallelism":"4"}' -X POST 
http://localhost:8081/v1/jars/43e844ef-382f-45c3-aa2f-00549acd961e_App.jar/run

PATCH /jobs/<jobId>

POST /jobs/<jobId>/savepoints

curl -d '{"target-directory":"file:///savepoints", "cancel-job":"false"}' 
-X POST http://localhost:8081/v1/jobs/e99cdb41b422631c8ee2218caa6af1cc/savepoints
{"request-id":"ebde90836b8b9dc2da90e9e7655f4179"}

curl -X GET http://localhost:8081/v1/jobs/e99cdb41b422631c8ee2218caa6af1cc/
savepoints/ebde90836b8b9dc2da90e9e7655f4179
{"status":{"id":"COMPLETED"} 
"operation":{"location":"file:///savepoints/savepoint-e99cdb-34410597dec0"}}

POST /savepoint-disposal

curl -d '{"savepoint-path":"file:///savepoints/savepoint-e99cdb-34410597"}'
-X POST http://localhost:8081/v1/savepoint-disposal
{"request-id":"217a4ffe935ceac2c281bdded76729d6"}

PATCH /jobs/<jobID>/rescaling

curl -X PATCH
http://localhost:8081/v1/jobs/129ced9aacf1618ebca0ba81a4b222c6/rescaling
?parallelism=16
{"request-id":"39584c2f742c3594776653f27833e3eb"}

Bundling and Deploying Applications in Containers

So far we have explained how to start an application on a running Flink cluster. This is what we call the framework style of deploying applications. In “Application Deployment”, we briefly explained an alternative—the library mode that does not require a running Flink cluster to submit a job.

In library mode, the application is bundled into a Docker image that also includes the required Flink binaries. The image can be started in two ways—as a JobMaster container or a TaskManager container. When the image is deployed as a JobMaster, the container starts a Flink master process that immediately picks up the bundled application to start it. A TaskManager container registers itself at the JobMaster and offers its processing slots. As soon as enough slots become available, the JobMaster container deploys the application for execution.

The library style of running Flink applications resembles the deployment of microservices in a containerized environment. When being deployed on a container orchestration framework, such as Kubernetes, the framework restarts failed containers. In this section, we describe how to build a job-specific Docker image and how to deploy a library-style bundled application on Kubernetes.

cd ./flink-container/docker
./build.sh 
    --from-archive <path-to-Flink-1.7.1-archive> 
    --job-jar <path-to-example-apps-JAR-file> 
    --image-name flink-book-apps

FLINK_DOCKER_IMAGE_NAME=flink-book-jobs 
  FLINK_JOB=io.github.streamingwithflink.chapter1.AverageSensorReadings 
  DEFAULT_PARALLELISM=3 
  docker-compose up -d

./flink-container/kubernetes

kubectl create -f job-cluster-service.yaml
kubectl create -f job-cluster-job.yaml
kubectl create -f task-manager-deployment.yaml

eval $(minikube docker-env)

Controlling Task Chaining

Task chaining fuses the parallel tasks of two or more operators into a single task that is executed by a single thread. The fused tasks exchange records by method calls and thus with basically no communication costs. Since task chaining improves the performance of most applications, it is enabled by default in Flink.

However, certain applications might not benefit from task chaining. One reason is to break a chain of expensive functions in order to execute them on different processing slots. You can completely disable task chaining for an application via the StreamExecutionEnvironment:

StreamExecutionEnvironment.disableOperatorChaining()

In addition to disabling chaining for the whole application, you can also control the chaining behavior of individual operators. To disable chaining for a specific operator, you can call its disableChaining() method. This will prevent the tasks of the operator from being chained to preceding and succeeding tasks (Example 10-1).

val input: DataStream[X] = ...
val result: DataStream[Y] = input
  .filter(new Filter1())
  .map(new Map1())
  // disable chaining for Map2
  .map(new Map2()).disableChaining()
  .filter(new Filter2())

The code in Example 10-1 results in three tasks—a chained task for Filter1 and Map1, an individual task for Map2, and a task for Filter2, which is not allowed to be chained to Map2.

It is also possible to start a new chain with an operator by calling its startNewChain() method (Example 10-2). The tasks of the operator will not be chained to preceding tasks but will be chained to succeeding tasks if the requirements for chaining are met.

val input: DataStream[X] = ...
val result: DataStream[Y] = input
  .filter(new Filter1())
  .map(new Map1())
  // start a new chain for Map2 and Filter2
  .map(new Map2()).startNewChain()
  .filter(new Filter2())

In Example 10-2 two chained tasks are created: one task for Filter1 and Map1 and another task for Map2 and Filter2. Note that the new chained task starts with the operator on which the startNewChain() method is called—Map2 in our example.

Defining Slot-Sharing Groups

Flink’s default task scheduling strategy assigns a complete slice of a program—up to one task of each operator of an application to a single processing slot.² Depending on the complexity of the application and the computational costs of the operators, this default strategy can overload a processing slot. Flink’s mechanism to manually control the assignment of tasks to slots is slot-sharing groups.

Each operator is a member of a slot-sharing group. All tasks of operators that are members of the same slot-sharing group are processed by the same slots. Within a slot-sharing group, the tasks are assigned to slots as described in “Task Execution”—each slot processes up to one task of each operator that is a member. Hence, a slot-sharing group requires as many processing slots as the maximum parallelism of its operators. Tasks of operators that are in different slot-sharing groups are not executed by the same slots.

By default, each operator is in the "default" slot-sharing group. For each operator, you can explicitly specify its slot-sharing group with the slotSharingGroup(String) method. An operator inherits the slot-sharing group of its input operators if they are all members of the same group. If the input operators are in different groups, the operator is in the "default" group. Example 10-3 shows how to specify slot-sharing groups in a Flink DataStream application.

// slot-sharing group "green"
val a: DataStream[A] = env.createInput(...)
  .slotSharingGroup("green")
  .setParallelism(4)
val b: DataStream[B] = a.map(...)
  // slot-sharing group "green" is inherited from a
  .setParallelism(4)

// slot-sharing group "yellow"
val c: DataStream[C] = env.createInput(...)
  .slotSharingGroup("yellow")
  .setParallelism(2)

// slot-sharing group "blue"
val d: DataStream[D] = b.connect(c.broadcast(...)).process(...)
  .slotSharingGroup("blue")
  .setParallelism(4)
val e = d.addSink()
  // slot-sharing group "blue" is inherited from d
  .setParallelism(2)

The application in Example 10-3 consists of five operators, two sources, two intermediate operators, and a sink operator. The operators are assigned to three slot-sharing groups: green, yellow, and blue. Figure 10-1 shows the JobGraph of the application and how its tasks are mapped to processing slots.

Figure 10-1. Controlling task scheduling with slot-sharing groups

The application requires 10 processing slots. The blue and green slot-sharing groups require four slots each due to the maximum parallelism of their assigned operators. The yellow slot-sharing group requires two slots.

Configuring Checkpointing

When you enable checkpointing for an application, you have to specify the checkpointing interval—the interval in which the JobManager will initiate checkpoints at the sources of the application.

Checkpoints are enabled on the StreamExecutionEnvironment:

val env: StreamExecutionEnvironment = ???

// enable checkpointing with an interval of 10 seconds.
env.enableCheckpointing(10000);

Further options to configure the checkpointing behavior are provided by the CheckpointConfig, which can be obtained from the StreamExecutionEnvironment:

// get the CheckpointConfig from the StreamExecutionEnvironment
val cpConfig: CheckpointConfig = env.getCheckpointConfig

By default, Flink creates checkpoints to guarantee exactly-once state consistency. However, it can also be configured to provide at-least-once guarantees:

// set mode to at-least-once
cpConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

Depending on the characteristics of an application, the size of its state, and the state backend and its configuration, a checkpoint can take up to a few minutes. Moreover, the size of the state can grow and shrink over time, perhaps due to long-running windows. Hence, it is not uncommon for a checkpoint to take more time than the configured checkpointing interval. By default, Flink allows only one checkpoint to be in progress at a time to avoid checkpointing takeing away too many resources needed for regular processing. If—according to the configured checkpointing interval—a checkpoint needs to be started, but there is another checkpoint in progress, the second checkpoint will be put on hold until the first checkpoint completes.

If many or all checkpoints take longer than the checkpointing interval, this behavior might not be optimal for two reasons. First, it means that the regular data processing of the application will always compete for resources with the concurrent checkpointing. Hence, its processing slows down and it might not be able to make enough progress to keep up with the incoming data. Second, a checkpoint may be delayed because we need to wait for another checkpoint to complete results in a lower checkpointing interval, leading to longer catch-up processing during recovery. Flink provides parameters to address these situations.

To ensure an application can make enough progress, you can configure a minimum pause between checkpoints. If you configure the minimum pause to be 30 seconds, then no new checkpoint will be started within the first 30 seconds after a checkpoint completed. This also means the effective checkpointing interval is at least 30 seconds and there is at most one checkpoint happening at the same time.

// make sure we process at least 30s without checkpointing
cpConfig.setMinPauseBetweenCheckpoints(30000);

In certain situations you might want to ensure that checkpoints are taken in the configured checkpointing interval even though a checkpoint takes longer than the interval. One example would be when checkpoints take a long time but do not consume much resources; for example, due to operations with high-latency calls to external systems. In this case you can configure the maximum number of concurrent checkpoints.

// allow three checkpoint to be in progress at the same time 
cpConfig.setMaxConcurrentCheckpoints(3);

To avoid long-running checkpoints, you can configure a timeout interval after which a checkpoint is canceled. By default, checkpoints are canceled after 10 minutes.

// checkpoints have to complete within five minutes, or are discarded
cpConfig.setCheckpointTimeout(300000);

Finally, you might also want to configure what happens if a checkpoint fails. By default, a failing checkpoint causes an exception that results in an application restart. You can disable this behavior and let the application continue after a checkpointing error.

// do not fail the job on a checkpointing error
cpConfig.setFailOnCheckpointingErrors(false);

val env: StreamExecutionEnvironment = ???

// enable checkpoint compression
env.getConfig.setUseSnapshotCompression(true)

// Enable externalized checkpoints
cpConfig.enableExternalizedCheckpoints(
  ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

Configuring State Backends

The state backend of an application is responsible for maintaining the local state, performing checkpoints and savepoints, and recovering the application state after a failure. Hence, the choice and configuration of the application’s state backend has a large impact on the performance of the checkpoints. The individual state backends are described in more detail in “Choosing a State Backend”.

The default state backend of an application is MemoryStateBackend. Since it holds all state in memory and checkpoints are completely stored in the volatile and JVM-size limited JobManager heap storage, it is not recommended for production environments. However, it serves well for locally developing Flink applications. “Checkpointing and State Backends” describes how you can configure a default state backend of a Flink cluster.

You can also explicitly choose the state backend of an application:

val env: StreamExecutionEnvironment = ???

// create and configure state backend of your choice
val stateBackend: StateBackend = ???
// set state backend
env.setStateBackend(stateBackend)

The different state backends can be created with minimum settings as shown in the following. MemoryStateBackend does not require any parameters. However, there are constructors that take parameters to enable or disable asynchronous checkpointing (enabled by default) and limit the size of state (5 MB by default):

// create a MemoryStateBackend
val memBackend = new MemoryStateBackend()

FsStateBackend only requires a path to define the storage location for checkpoints. There are also constructor variants to enable or disable asynchronous checkpointing (enabled by default):

// create a FsStateBackend that checkpoints to the /tmp/ckp folder
val fsBackend = new FsStateBackend("file:///tmp/ckp", true)

RocksDBStateBackend only requires a path to define the storage location for checkpoints and takes an optional parameter to enable incremental checkpoints (disabled by default). RocksDBStateBackend is always writing checkpoints asynchronously:

// create a RocksDBStateBackend that writes incremental checkpoints 
// to the /tmp/ckp folder
val rocksBackend = new RocksDBStateBackend("file:///tmp/ckp", true)

In “Checkpointing and State Backends”, we discussed the configuration options for state backends. You can, of course, also configure the state backend in your application, overriding the default values or cluster-wide configuration. For that you have to create a new backend object by passing a Configuration object to your state backend. See “Checkpointing and State Backends” for a description of the available configuration options:

// all of Flink's built-in backends are configurable
val backend: ConfigurableStateBackend = ??? 

// create configuration and set options
val sbConfig = new Configuration()
sbConfig.setBoolean("state.backend.async", true)
sbConfig.setString("state.savepoints.dir", "file:///tmp/svp")

// create a configured copy of the backend
val configuredBackend = backend.configure(sbConfig)

Since RocksDB is an external component, it brings its own set of tuning parameters that can also be tweaked for your application. By default, RocksDB is optimized for SSD storage and does not provide great performance if state is stored on spinning disks. Flink provides a few predefined settings to improve the performance for common hardware setups. See the documentation to learn more about the available settings. You can apply predefined options to RocksDBStateBackend as follows:

val backend: RocksDBStateBackend = ???

// set predefined options for spinning disk storage
backend.setPredefinedOptions(PredefinedOptions.SPINNING_DISK_OPTIMIZED)

Configuring Recovery

When a checkpointed application fails, it will be restarted by bringing up its tasks, recovering their states, including the reading offsets of the source tasks, and continuing the processing. Right after the application was restarted it is in a catch-up phase. Since the application’s source tasks were reset to an earlier input position, it processes data that it processed before the failure and data that accumulated while the application was down.

To be able to catch up with the stream—reach its tail—the application must process the accumulated data at a higher rate than new data is arriving. While the application is catching up, the processing latency—the time at which input is available until it is actually processed—increases.

Consequently, an application needs enough spare resources for the catch-up phase after the application was restarted to successfully resume its regular processing. This means an application should not run close to 100% resource consumption during regular processing. The more resources available for recovery, the faster the catch-up phase completes and the faster processing latencies go back to normal.

Besides resource considerations for the recovery, there are two other recovery-related topics we will discuss: restart strategies and local recovery.

val env = StreamExecutionEnvironment.getExecutionEnvironment

env.setRestartStrategy(
  RestartStrategies.fixedDelayRestart(
    5,                            // number of restart attempts
    Time.of(30, TimeUnit.SECONDS) // delay between attempts
))

Flink Web UI

The simplest way to get an overview of your Flink cluster as well as a glimpse of what your jobs are doing internally is to use Flink’s Web UI. You can access the dashboard by visiting http://<jobmanager-hostname>:8081.

On the home screen, you will see an overview of your cluster configuration including the number of TaskManagers, number of configured and available task slots, and running and completed jobs. Figure 10-2 shows an instance of the dashboard home screen. The menu on the left links to more detailed information on jobs and configuration parameters and also allows job submission by uploading a JAR.

Figure 10-2. Apache Flink Web UI home screen

If you click on a running job, you can get a quick glimpse of running statistics per task or subtask as shown in Figure 10-3. You can inspect the duration, bytes, and records exchanged, and aggregate those per TaskManager if you prefer.

Figure 10-3. Statistics for a running job

If you click on the Task Metrics tab, you can select more metrics from a dropdown menu, as shown in Figure 10-4. These include more fine-grained statistics about your tasks, such as buffer usage, watermarks, and input/output rates.

Figure 10-4. Selecting metrics to plot

Figure 10-5 shows how selected metrics are shown as continuously updated charts.

Figure 10-5. Real-time metric plots

The Checkpoints tab (Figure 10-3) displays statistics about previous and current checkpoints. Under Overview you can see how many checkpoints have been triggered, are in progress, have completed successfully, or have failed. If you click on the History view, you can retrieve more fine-grained information, such as the status, trigger time, state size, and how many bytes were buffered during the checkpoint’s alignment phase. The Summary view aggregates checkpoint statistics and provides minimum, maximum, and average values over all completed checkpoints. Finally, under Configuration, you can inspect the configuration properties of checkpoints, such as the interval and the timeout values set.

Similarly, the Back Pressure tab displays back-pressure statistics per operator and subtask. If you click on a row, you trigger back-pressure sampling and you will see the message Sampling in progress... for about five seconds. Once sampling is complete, you will see the back-pressure status in the second column. Back-pressured tasks will display a HIGH sign; otherwise you should see a nice green OK message displayed.

Metric System

When running a data processing system such as Flink in production, it is essential to monitor its behavior to be able to discover and diagnose the cause of performance degradations. Flink collects several system and application metrics by default. Metrics are gathered per operator, TaskManager, or JobManager. Here we describe some of the most commonly used metrics and refer you to Flink’s documentation for a full list of available metrics.

Categories include CPU utilization, memory used, number of active threads, garbage collection statistics, network metrics such as the number of queued input/output buffers, cluster-wide metrics such as the number or running jobs and available resources, job metrics including runtime, the number of retries and checkpointing information, I/O statistics including the number of record exchanges locally and remotely, watermark information, connector-specific metrics, etc.

class PositiveFilter extends RichFilterFunction[Int] {

  @transient private var counter: Counter = _

  override def open(parameters: Configuration): Unit = {
    counter = getRuntimeContext
      .getMetricGroup
      .counter("droppedElements")
  }

  override def filter(value: Int): Boolean = {
    if (value > 0) {
      true
    }
    else {
      counter.inc()
      false
    }
  }
}

counter = getRuntimeContext
  .getMetricGroup
  .addGroup("MyMetrics")
  .counter("myCounter")

metrics.reporters: my_reporter
Metrics.reporter.my_jmx_reporter.class: org.apache.flink.metrics.jmx.JMXReporter
metrics.reporter.my_jmx_reporter.port: 9020-9040

Monitoring Latency

Latency is probably one of the first metrics you want to monitor to assess the performance characteristics of your streaming job. At the same time, it is also one of the trickiest metrics to define in a distributed streaming engine with rich semantics such as Flink. In “Latency”, we defined latency broadly as the time it takes to process an event. You can imagine how a precise implementation of this definition can get problematic in practice if we try to track the latency per event in a high-rate streaming job with a complex dataflow. Considering window operators complicate latency tracking even further, if an event contributes to several windows, do we need to report the latency of the first invocation or do we need to wait until we evaluate all windows an event might belong to? And what if a window triggers multiple times?

Flink follows a simple and low-overhead approach to provide useful latency metric measurements. Instead of trying to strictly measure latency for each and every event, it approximates latency by periodically emitting a special record at the sources and allowing users to track how long it takes for this record to arrive at the sinks. This special record is called a latency marker, and it bears a timestamp indicating when it was emitted.

To enable latency tracking, you need to configure how often latency markers are emitted from the sources. You can do this by setting the latencyTrackingInterval in the ExecutionConfig as shown here:

env.getConfig.setLatencyTrackingInterval(500L)

The interval is specified in milliseconds. Upon receiving a latency marker, all operators except sinks forward it downstream. Latency markers use the same dataflow channels and queues as normal stream records, thus their tracked latency reflects the time records wait to be processed. However, they do not measure the time it takes for records to be processed or the time that records wait in state until they are processed.

Operators keep latency statistics in a latency gauge that contains min, max, and mean values, as well as 50, 95, and 99 percentile values. Sink operators keep statistics on latency markers received per parallel source instance, thus checking the latency marker at sinks can be used to approximate how long it takes for records to traverse the dataflow. If you would like to customize the handling the latency marker at operators, you can override the processLatencyMarker() method and retrieve the relevant information using the LatencyMarker’s methods getMarkedTime(), getVertexId(), and getSubTaskIndex().