flink-kubernetes-operator 官方文档中给出的 application 模式 demo

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  namespace: default
  name: basic-example
spec:
  image: flink:1.16
  flinkVersion: v1_16
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
  serviceAccount: flink
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    parallelism: 2
    upgradeMode: stateless
    state: running

• 不难看出 application 模式 是使用local 方式管理flink jar 应用程序 (local:///opt/flink/examples/streaming/StateMachineExample.jar) 这就要求使用者自定义flink 镜像

Flink on-k8s operator 模式运行flink 使用dockerfile 方式进行构建

• 在编写dockerfile 之前使用 docker images 看下是否拉取了flink 基础镜像 如果没有执行

  docker pull  flink:1.15.3-scala_2.12-java8

第一步： 编写dockerfile

FROM   flink:1.15.3-scala_2.12-java8
USER root

ENV FLINK_HOME=/opt/flink
ENV PATH=$FLINK_HOME/bin:$PATH

#flink添加到根目录

RUN   mkdir $FLINK_HOME/userlib
ADD  flink-cdc-demo-1.0-jar-with-dependencies.jar $FLINK_HOME/userlib

#设置code文件夹的工作目录
EXPOSE 6123 8081

• 第二步：构建镜像到远程镜像仓库 (your_docker_repository_url 为镜像仓库地址 flink 为该仓库中的目录,flink_application 表示镜像名称)

    
    docker build -t   your_docker_repository_url/flink/flink_application . 

• 第三步: 上传构建好的镜像到远程仓库

  docker push  your_docker_repository_url/flink/flink_application
  

编写application job

• 通过image 标签来指定 自定义的flink 镜像

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: flink-application-1
  namespace: your_namespace
spec:
  mode: native
  image: www.abc.cn/flink/flink_application:1.15.3-scala_2.12-java8
  flinkVersion: v1_15
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "6"
    state.checkpoints.dir: file:///tmp/checkpoint
  serviceAccount: flink
  jobManager:
    resource:
      memory: "1024m"
      cpu: 1
  taskManager:
    resource:
      memory: "4096m"
      cpu: 1
  job:
    jarURI: local:///opt/flink/userlib/flink-cdc-demo-1.0-jar-with-dependencies.jar
    parallelism: 1
    upgradeMode: stateless
    entryClass: MysqlSinkPulsar
    args: ["params1=xxxxxx","params2="]

最后运行application flink

    kubectl  apply -f  your_FlinkDeployment.yaml 

• 备注： 亲测有效