Use Docker to configure flink1 12 development test environment

Brief introduction

Docker is now an indispensable tool in the development community. We can quickly test and package our applications with docker. At the same time, using containers, we can easily realize the deployment and configuration of CI and CD in software development. At present, the official basic images of Flink can be downloaded from the docker hub. However, the docker hub imposed a speed limit on tourists some time ago, so when you encounter a slow download speed, please log in with docker login first, and then use docker pull to pull the image file.

Using Docker's image, we can build Session mode or Application mode Flink clusters. Now I will introduce you step by step.

Start a Flink session cluster in the docker environment

Each session cluster can run multiple Flink tasks at the same time. Of course, you need enough resources. Flink cluster needs to be started before running tasks. To run a Flink session cluster, you need to run a JobManager container and multiple TaskManager containers. The communication between JobManager and TaskManager adopts rpc, but in the docker environment, we first need to create a Flink dedicated network. Of course, if docker compose is used, it will be created by default.

If you don't have an environment, look here for a quick [build docker environment]( https://blog.csdn.net/lihengzkj/article/details/116136604)

1. First, create a network for Flink's own use. Of course, you can also use the default network without creating it: docker network create Flink test network
2. Run a JobManager using docker:

   docker run -d \
   --name jobnamager \
   --network flink-test-network \
   -- publish 8081:8081
   --env jobmanager.rpc.address=jobmanager \
   flink:1.12.2-scala_2.11-java11 jobmanager
   

3. Run a task manager using docker

   docker run -d \
   --name taskmanager \
   --network flink-test-network \
   --env jobmanager.rpc.address=jobmanager \
   flink:1.12.2-scala_2.11-java11 taskmanager
   

4. visit: http://localhost:8081
5. Test: upload Jar package test through UI or run test code through client.
6. Stop cluster:
   1. Find the id of the container: docker ps | grep -E "(jobmanager|taskmanager)"
   2. Delete container: docker rm -f container id

Deployment mode

Flink's image contains a regular configuration and entry script. We can start the Flink cluster through the entry script.

• The deployment modes of JobManager are:
  • Seesion cluster mode
  • Application cluster mode
• Task manager adapts to any mode

What are the specific differences between the above two modes? Please see my introduction: [introduction to Flink1.12.3 deployment mode]( https://blog.csdn.net/lihengzkj/article/details/116474617)

Application mode on Docker

In this mode, Flink only runs one job in the whole cluster, that is, when the cluster starts, the only job starts.

This mode requires us to put our implemented jar package into the image or container in advance. At the same time, we should pay attention to putting the corresponding dependencies in. The best way is to make all dependencies into one package. Here are the specific steps:

Local test environment

• Put your jar package, which is called artifact for Flink, into / opt/flink/artifacts through mount or volume
• Start JobManager in Application cluster mode
• Start one or more task managers

In the docker environment, the following commands can be used:

• First, create a proprietary network:
  • docker network creat flink-network
• Mount our jar package and start a jobmanager: jobid, savepoint and arguments in [] are all optional. The program can fill them in as needed

    docker run \
      --mount type=bind,src=/host/path/to/job/xxx1.jar,target=/opt/flink/artifacts/xxx1.jar \
      --mount type=bind,src=/host/path/to/job/xxx2.jar,target=/opt/flink/artifacts/xxx2.jar \
      --rm \
      --env jobmanager.rpc.address=jobmanager \
      --name=jobmanager \
      --network flink-network \
      flink:1.12.2-scala_2.11-java11 standalone-job \
      --job-classname com.job.ClassName \
      [--job-id <job id>] \
      [--fromSavepoint /path/to/savepoint [--allowNonRestoredState]] \
      [job arguments]
  

• Mount our jar package and start a taskmanager:

    docker run \
      --mount type=bind,src=/host/path/to/job/xxx1.jar,target=/opt/flink/artifacts/xxx1.jar \
      --mount type=bind,src=/host/path/to/job/xxx2.jar,target=/opt/flink/artifacts/xxx2.jar \
      --env jobmanager.rpc.address=jobmanager \
      flink:1.12.2-scala_2.11-java11 taskmanager
  

CI test environment

If your development test is on gitlab, you can use Dockerfile to automatically create local images of Flink's jobmanager and taskmanager

• If we have a project on the Dockerfile of Java. It should be noted here that the root directory of the project can usually be packaged directly with the maven command.

FROM maven:xxx

RUN mvn clean install/package

FROM flink:1.12.2-scala_2.11-java11

ADD ./target/xxx.jar /opt/flink/artifacts/xxx.jar

• Package into image: docker build - t flick: local-1.0
• Execution:
  • Or run a jobmanager according to the above application mode:

    docker run \
      flink:local-1.0 standalone-job \
      --job-classname com.job.ClassName
  

  • Then run a taskmanager: docker run flick: local-1.0 taskmanager

session cluster used in docker compose environment

Generally speaking, we need a stable session cluster environment for testing. We will provide you with a docker compose environment. As long as you run under docker compose, you will start a Jobmanager and a taskmanager cluster. Of course, you can copy multiple taskmanagers in yaml file as long as your resources are sufficient.

version: "3.8"
services:
  flink:
    image: flink:1.12.2-scala_2.11-java11
    restart: unless-stopped
    command: jobmanager
    environment:
      JOB_MANAGER_RPC_ADDRESS: flink
      FLINK_PROPERTIES: |
        cluster.evenly-spread-out-slots: true
        restart-strategy: failure-rate
        restart-strategy.failure-rate.delay: 10 s
        restart-strategy.failure-rate.failure-rate-interval: 10
        restart-strategy.failure-rate.max-failures-per-interval: 2
    ports: [8081:8081]
    volumes:
      - /opt/flink/log
      - /tmp
      
  flink-worker:
    image: flink:1.12.2-scala_2.11-java11
    restart: unless-stopped
    command: taskmanager
    environment:
      JOB_MANAGER_RPC_ADDRESS: flink
    depends_on: [flink]
    volumes:
      - /opt/flink/log
      - /tmp