Flink Operations Playground

    In this playground, you will learn how to manage and run Flink Jobs. You will see how to deploy and monitor an application, experience how Flink recovers from Job failure, and perform everyday operational tasks like upgrades and rescaling.

    This playground consists of a long living Flink Session Cluster and a Kafka Cluster.

    A Flink Cluster always consists of a and one or more Flink TaskManagers. The JobManager is responsible for handling submissions, the supervision of Jobs as well as resource management. The Flink TaskManagers are the worker processes and are responsible for the execution of the actual Tasks which make up a Flink Job. In this playground you will start with a single TaskManager, but scale out to more TaskManagers later. Additionally, this playground comes with a dedicated client container, which we use to submit the Flink Job initially and to perform various operational tasks later on. The client container is not needed by the Flink Cluster itself but only included for ease of use.

    The Kafka Cluster consists of a Zookeeper server and a Kafka Broker.

    When the playground is started a Flink Job called Flink Event Count will be submitted to the JobManager. Additionally, two Kafka Topics input and output are created.

    Click Event Count Example

    The Job consumes s from the input topic, each with a timestamp and a page. The events are then keyed by page and counted in 15 second . The results are written to the output topic.

    There are six different pages and we generate 1000 click events per page and 15 seconds. Hence, the output of the Flink job should show 1000 views per page and window.

    Back to top

    The playground environment is set up in just a few steps. We will walk you through the necessary commands and show how to validate that everything is running correctly.

    We assume that you have (1.12+) and docker-compose (2.1+) installed on your machine.

    The required configuration files are available in the repository. First checkout the code and build the docker image:

    Then before starting the playground, create the checkpoint and savepoint directories on the Docker host machine (these volumes are mounted by the jobmanager and taskmanager, as specified in docker-compose.yaml):

    1. mkdir -p /tmp/flink-checkpoints-directory
    2. mkdir -p /tmp/flink-savepoints-directory

    Then start the playground:

    1. docker-compose up -d

    Afterwards, you can inspect the running Docker containers with the following command:

    1. docker-compose ps
    3.                     Name                                  Command               State                   Ports                
    4. -----------------------------------------------------------------------------------------------------------------------------
    5. operations-playground_clickevent-generator_1   /docker-entrypoint.sh java ...   Up       6123/tcp, 8081/tcp                  
    6. operations-playground_client_1                 /docker-entrypoint.sh flin ...   Exit 0                                       
    7. operations-playground_jobmanager_1             /docker-entrypoint.sh jobm ...   Up       6123/tcp, 0.0.0.0:8081->8081/tcp    
    8. operations-playground_kafka_1                  start-kafka.sh                   Up       0.0.0.0:9094->9094/tcp              
    9. operations-playground_taskmanager_1            /docker-entrypoint.sh task ...   Up       6123/tcp, 8081/tcp                  
    10. operations-playground_zookeeper_1              /bin/sh -c /usr/sbin/sshd  ...   Up       2181/tcp, 22/tcp, 2888/tcp, 3888/tcp

    This indicates that the client container has successfully submitted the Flink Job (Exit 0) and all cluster components as well as the data generator are running (Up).

    You can stop the playground environment by calling:

    1. docker-compose down -v

    There are many things you can try and check out in this playground. In the following two sections we will show you how to interact with the Flink Cluster and demonstrate some of Flink’s key features.

    The most natural starting point to observe your Flink Cluster is the WebUI exposed under http://localhost:8081. If everything went well, you’ll see that the cluster initially consists of one TaskManager and executes a Job called Click Event Count.

    The Flink WebUI contains a lot of useful and interesting information about your Flink Cluster and its Jobs (JobGraph, Metrics, Checkpointing Statistics, TaskManager Status,…).

    Logs

    JobManager

    The JobManager logs can be tailed via docker-compose.

    1. docker-compose logs -f jobmanager

    After the initial startup you should mainly see log messages for every checkpoint completion.

    TaskManager

    The TaskManager log can be tailed in the same way.

    1. docker-compose logs -f taskmanager

    After the initial startup you should mainly see log messages for every checkpoint completion.

    The can be used from within the client container. For example, to print the help message of the Flink CLI you can run

    1. docker-compose run --no-deps client flink --help

    The Flink REST API is exposed via localhost:8081 on the host or via jobmanager:8081 from the client container, e.g. to list all currently running jobs, you can run:

    1. curl localhost:8081/jobs

    Kafka Topics

    You can look at the records that are written to the Kafka Topics by running

    1. //input topic (1000 records/s)
    2. docker-compose exec kafka kafka-console-consumer.sh \
    3.   --bootstrap-server localhost:9092 --topic input
    5. //output topic (24 records/min)
    6. docker-compose exec kafka kafka-console-consumer.sh \
    7.   --bootstrap-server localhost:9092 --topic output

    Back to top

    Listing Running Jobs

    CLI

    Command

    1. docker-compose run --no-deps client flink list

    Expected Output

    1. Waiting for response...
    2. ------------------ Running/Restarting Jobs -------------------
    3. 16.07.2019 16:37:55 : <job-id> : Click Event Count (RUNNING)
    4. --------------------------------------------------------------
    5. No scheduled jobs.

    REST API

    Request

    1. curl localhost:8081/jobs

    Expected Response (pretty-printed)

    The JobID is assigned to a Job upon submission and is needed to perform actions on the Job via the CLI or REST API.

    Flink provides exactly-once processing guarantees under (partial) failure. In this playground you can observe and - to some extent - verify this behavior.

    Step 1: Observing the Output

    As described , the events in this playground are generated such that each window contains exactly one thousand records. So, in order to verify that Flink successfully recovers from a TaskManager failure without data loss or duplication you can tail the output topic and check that - after recovery - all windows are present and the count is correct.

    For this, start reading from the output topic and leave this command running until after recovery (Step 3).

    1. docker-compose exec kafka kafka-console-consumer.sh \
    2.   --bootstrap-server localhost:9092 --topic output

    Step 2: Introducing a Fault

    In order to simulate a partial failure you can kill a TaskManager. In a production setup, this could correspond to a loss of the TaskManager process, the TaskManager machine or simply a transient exception being thrown from the framework or user code (e.g. due to the temporary unavailability of an external resource).

    1. docker-compose kill taskmanager

    After a few seconds, the JobManager will notice the loss of the TaskManager, cancel the affected Job, and immediately resubmit it for recovery. When the Job gets restarted, its tasks remain in the SCHEDULED state, which is indicated by the purple colored squares (see screenshot below).

    Playground Flink WebUI

    At this point, the tasks of the Job cannot move from the SCHEDULED state to RUNNING because there are no resources (TaskSlots provided by TaskManagers) to the run the tasks. Until a new TaskManager becomes available, the Job will go through a cycle of cancellations and resubmissions.

    In the meantime, the data generator keeps pushing ClickEvents into the input topic. This is similar to a real production setup where data is produced while the Job to process it is down.

    Step 3: Recovery

    Once you restart the TaskManager, it reconnects to the JobManager.

    1. docker-compose up -d taskmanager

    When the JobManager is notified about the new TaskManager, it schedules the tasks of the recovering Job to the newly available TaskSlots. Upon restart, the tasks recover their state from the last successful checkpoint that was taken before the failure and switch to the RUNNING state.

    The Job will quickly process the full backlog of input events (accumulated during the outage) from Kafka and produce output at a much higher rate (> 24 records/minute) until it reaches the head of the stream. In the output you will see that all keys (pages) are present for all time windows and that every count is exactly one thousand. Since we are using the in its “at-least-once” mode, there is a chance that you will see some duplicate output records.

    Note: Most production setups rely on a resource manager (Kubernetes, Yarn) to automatically restart failed processes.

    Upgrading & Rescaling a Job

    Upgrading a Flink Job always involves two steps: First, the Flink Job is gracefully stopped with a . A Savepoint is a consistent snapshot of the complete application state at a well-defined, globally consistent point in time (similar to a checkpoint). Second, the upgraded Flink Job is started from the Savepoint. In this context “upgrade” can mean different things including the following:

    • An upgrade to the configuration (incl. the parallelism of the Job)
    • An upgrade to the topology of the Job (added/removed Operators)
    • An upgrade to the user-defined functions of the Job

    Before starting with the upgrade you might want to start tailing the output topic, in order to observe that no data is lost or corrupted in the course the upgrade.

    1. docker-compose exec kafka kafka-console-consumer.sh \
    2.   --bootstrap-server localhost:9092 --topic output

    Step 1: Stopping the Job

    To gracefully stop the Job, you need to use the “stop” command of either the CLI or the REST API. For this you will need the JobID of the Job, which you can obtain by or from the WebUI. With the JobID you can proceed to stopping the Job:

    CLI

    Command

    1. docker-compose run --no-deps client flink stop <job-id>

    Expected Output

    1. Suspending job "<job-id>" with a savepoint.
    2. Savepoint completed. Path: file:<savepoint-path>

    The Savepoint has been stored to the state.savepoints.dir configured in the flink-conf.yaml, which is mounted under /tmp/flink-savepoints-directory/ on your local machine. You will need the path to this Savepoint in the next step.

    REST API

    Request

    1. # triggering stop
    2. curl -X POST localhost:8081/jobs/<job-id>/stop -d '{"drain": false}'

    Expected Response (pretty-printed)

    1. {
    2.   "request-id": "<trigger-id>"
    3. }

    Request

    1. # check status of stop action and retrieve savepoint path
    2.  curl localhost:8081/jobs/<job-id>/savepoints/<trigger-id>

    Expected Response (pretty-printed)

    1. {
    2.   "status": {
    3.     "id": "COMPLETED"
    4.   },
    5.   "operation": {
    6.     "location": "<savepoint-path>"
    7.   }
    8. }

    Step 2a: Restart Job without Changes

    CLI

    Command

    1. docker-compose run --no-deps client flink run -s <savepoint-path> \
    2.   -d /opt/ClickCountJob.jar \
    3.   --bootstrap.servers kafka:9092 --checkpointing --event-time

    Expected Output

    1. Job has been submitted with JobID <job-id>

    REST API

    Request

    Expected Response (pretty-printed)

    1. {
    2.   "filename": "/tmp/flink-web-<uuid>/flink-web-upload/<jar-id>",
    3.   "status": "success"
    4. }

    Request

    1. # Submitting the Job
    2. curl -X POST http://localhost:8081/jars/<jar-id>/run \
    3.   -d '{"programArgs": "--bootstrap.servers kafka:9092 --checkpointing --event-time", "savepointPath": "<savepoint-path>"}'

    Expected Response (pretty-printed)

    1. {
    2.   "jobid": "<job-id>"
    3. }

    Once the Job is RUNNING again, you will see in the output Topic that records are produced at a higher rate while the Job is processing the backlog accumulated during the outage. Additionally, you will see that no data was lost during the upgrade: all windows are present with a count of exactly one thousand.

    Step 2b: Restart Job with a Different Parallelism (Rescaling)

    Alternatively, you could also rescale the Job from this Savepoint by passing a different parallelism during resubmission.

    CLI

    Command

    1. docker-compose run --no-deps client flink run -p 3 -s <savepoint-path> \
    2.   -d /opt/ClickCountJob.jar \
    3.   --bootstrap.servers kafka:9092 --checkpointing --event-time

    Expected Output

    1. Starting execution of program

    REST API

    Request

    1. # Uploading the JAR from the Client container
    2. docker-compose run --no-deps client curl -X POST -H "Expect:" \
    3.   -F "jarfile=@/opt/ClickCountJob.jar" http://jobmanager:8081/jars/upload

    Expected Response (pretty-printed)

    1. {
    2.   "filename": "/tmp/flink-web-<uuid>/flink-web-upload/<jar-id>",
    3.   "status": "success"

    Request

    1. # Submitting the Job
    2. curl -X POST http://localhost:8081/jars/<jar-id>/run \
    3.   -d '{"parallelism": 3, "programArgs": "--bootstrap.servers kafka:9092 --checkpointing --event-time", "savepointPath": "<savepoint-path>"}'

    Expected Response (pretty-printed

    1. {
    2.   "jobid": "<job-id>"
    3. }

    Now, the Job has been resubmitted, but it will not start as there are not enough TaskSlots to execute it with the increased parallelism (2 available, 3 needed). With

    1. docker-compose scale taskmanager=2

    you can add a second TaskManager with two TaskSlots to the Flink Cluster, which will automatically register with the JobManager. Shortly after adding the TaskManager the Job should start running again.

    Once the Job is “RUNNING” again, you will see in the output Topic that no data was lost during rescaling: all windows are present with a count of exactly one thousand.

    Querying the Metrics of a Job

    The JobManager exposes system and user via its REST API.

    The endpoint depends on the scope of these metrics. Metrics scoped to a Job can be listed via jobs/<job-id>/metrics. The actual value of a metric can be queried via the get query parameter.

    Request

    1. curl "localhost:8081/jobs/<jod-id>/metrics?get=lastCheckpointSize"

    Expected Response (pretty-printed; no placeholders)

    1. [
    2.   {
    3.     "id": "lastCheckpointSize",
    4.     "value": "9378"
    5.   }
    6. ]

    The REST API can not only be used to query metrics, but you can also retrieve detailed information about the status of a running Job.

    Request

    Expected Response (pretty-printed)

    1. {
    2.   "jid": "<job-id>",
    3.   "name": "Click Event Count",
    4.   "isStoppable": false,
    5.   "state": "RUNNING",
    6.   "start-time": 1564467066026,
    7.   "end-time": -1,
    8.   "duration": 374793,
    9.   "now": 1564467440819,
    10.   "timestamps": {
    11.     "CREATED": 1564467066026,
    12.     "FINISHED": 0,
    13.     "SUSPENDED": 0,
    14.     "FAILING": 0,
    15.     "CANCELLING": 0,
    16.     "CANCELED": 0,
    17.     "RECONCILING": 0,
    18.     "RUNNING": 1564467066126,
    19.     "FAILED": 0,
    20.     "RESTARTING": 0
    21.   },
    22.   "vertices": [
    23.     {
    24.       "id": "<vertex-id>",
    25.       "name": "ClickEvent Source",
    26.       "parallelism": 2,
    27.       "status": "RUNNING",
    28.       "start-time": 1564467066423,
    29.       "end-time": -1,
    30.       "duration": 374396,
    31.       "tasks": {
    32.         "CREATED": 0,
    33.         "FINISHED": 0,
    34.         "DEPLOYING": 0,
    35.         "RUNNING": 2,
    36.         "CANCELING": 0,
    37.         "FAILED": 0,
    38.         "CANCELED": 0,
    39.         "RECONCILING": 0,
    40.         "SCHEDULED": 0
    41.       },
    42.       "metrics": {
    43.         "read-bytes": 0,
    44.         "read-bytes-complete": true,
    45.         "write-bytes": 5033461,
    46.         "write-bytes-complete": true,
    47.         "read-records": 0,
    48.         "read-records-complete": true,
    49.         "write-records": 166351,
    50.         "write-records-complete": true
    51.       }
    52.     },
    53.     {
    54.       "id": "<vertex-id>",
    55.       "name": "ClickEvent Counter",
    56.       "parallelism": 2,
    57.       "status": "RUNNING",
    58.       "start-time": 1564467066469,
    59.       "end-time": -1,
    60.       "duration": 374350,
    61.       "tasks": {
    62.         "CREATED": 0,
    63.         "FINISHED": 0,
    64.         "DEPLOYING": 0,
    65.         "RUNNING": 2,
    66.         "CANCELING": 0,
    67.         "FAILED": 0,
    68.         "CANCELED": 0,
    69.         "RECONCILING": 0,
    70.         "SCHEDULED": 0
    71.       },
    72.       "metrics": {
    73.         "read-bytes": 5085332,
    74.         "read-bytes-complete": true,
    75.         "write-bytes": 316,
    76.         "write-bytes-complete": true,
    77.         "read-records-complete": true,
    78.         "write-records": 6,
    79.         "write-records-complete": true
    80.       }
    81.     },
    83.       "id": "<vertex-id>",
    84.       "name": "ClickEventStatistics Sink",
    85.       "parallelism": 2,
    86.       "status": "RUNNING",
    87.       "start-time": 1564467066476,
    88.       "end-time": -1,
    89.       "duration": 374343,
    90.       "tasks": {
    91.         "CREATED": 0,
    92.         "FINISHED": 0,
    93.         "DEPLOYING": 0,
    94.         "RUNNING": 2,
    95.         "CANCELING": 0,
    96.         "FAILED": 0,
    97.         "CANCELED": 0,
    98.         "RECONCILING": 0,
    99.         "SCHEDULED": 0
    100.       },
    101.       "metrics": {
    102.         "read-bytes": 20668,
    103.         "read-bytes-complete": true,
    104.         "write-bytes": 0,
    105.         "write-bytes-complete": true,
    106.         "read-records": 6,
    107.         "read-records-complete": true,
    108.         "write-records": 0,
    109.         "write-records-complete": true
    110.       }
    111.     }
    112.   ],
    113.   "status-counts": {
    114.     "CREATED": 0,
    115.     "FINISHED": 0,
    116.     "DEPLOYING": 0,
    117.     "RUNNING": 4,
    118.     "CANCELING": 0,
    119.     "FAILED": 0,
    120.     "CANCELED": 0,
    121.     "RECONCILING": 0,
    122.     "SCHEDULED": 0
    123.   },
    124.   "plan": {
    125.     "jid": "<job-id>",
    126.     "name": "Click Event Count",
    127.     "type": "STREAMING",
    128.     "nodes": [
    129.       {
    130.         "id": "<vertex-id>",
    131.         "parallelism": 2,
    132.         "operator": "",
    133.         "operator_strategy": "",
    134.         "description": "ClickEventStatistics Sink",
    135.         "inputs": [
    136.           {
    137.             "num": 0,
    138.             "id": "<vertex-id>",
    139.             "ship_strategy": "FORWARD",
    140.             "exchange": "pipelined_bounded"
    141.           }
    142.         ],
    143.         "optimizer_properties": {}
    144.       },
    145.       {
    146.         "id": "<vertex-id>",
    147.         "parallelism": 2,
    148.         "operator": "",
    149.         "operator_strategy": "",
    150.         "description": "ClickEvent Counter",
    151.         "inputs": [
    152.           {
    153.             "num": 0,
    154.             "id": "<vertex-id>",
    155.             "ship_strategy": "HASH",
    156.             "exchange": "pipelined_bounded"
    157.           }
    158.         ],
    159.         "optimizer_properties": {}
    160.       },
    161.       {
    162.         "id": "<vertex-id>",
    163.         "parallelism": 2,
    164.         "operator": "",
    165.         "operator_strategy": "",
    166.         "description": "ClickEvent Source",
    167.         "optimizer_properties": {}
    168.       }
    169.     ]
    170.   }
    171. }

    Please consult the REST API reference for a complete list of possible queries including how to query metrics of different scopes (e.g. TaskManager metrics);

    You might have noticed that the Click Event Count application was always started with --checkpointing and --event-time program arguments. By omitting these in the command of the client container in the docker-compose.yaml, you can change the behavior of the Job.

    • --checkpointing enables checkpoint, which is Flink’s fault-tolerance mechanism. If you run without it and go through , you should will see that data is actually lost.

    The Click Event Count application also has another option, turned off by default, that you can enable to explore the behavior of this job under backpressure. You can add this option in the command of the client container in docker-compose.yaml.