Flink on K8s简单实现

采用将程序Jar包上传的Flink镜像中的方式，使用Flink On K8s Standalone模式

一、编写Dockerflie

image.png

FROM flink:1.12-scala_2.11-java8
COPY ./target/xxx.jar /opt/flink/usrlib/xxx.jar
操作就是先拉取flink官方镜像，然后将本地jar包copy到镜像的usrlib目录下。

二、编写K8s资源描述文件

从Flink on Kubernetes 的架构如上图所示，Flink 任务在 Kubernetes 上运行的步骤有：

（1）首先往 Kubernetes 集群提交了资源描述文件后，会启动 Master 和 Worker 的 container。
（2）Master Container 中会启动 Flink Master Process，包含 Flink-Container ResourceManager、JobManager 和 Program Runner。
（3）Worker Container 会启动 TaskManager，并向负责资源管理的 ResourceManager 进行注册，注册完成之后，由 JobManager 将具体的任务分给 Worker Container ，再由 Container 去执行。
（4）需要说明的是，Master Container 与Worker Container是用一个镜像启动的，只是启动参数不一样，如下图所示，两个deployment文件的镜像是同一个。

具体描述文件详见附件，flink-jobmanager.yaml和flink-taskmanager.yaml，需根据实际情况更新image地址和启动的Class文件。

flink-jobmanager.yaml

apiVersion: batch/v1
kind: Job
metadata:
  name: xxx-flink-jobmanager
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: flink
        component: jobmanager
    spec:
      restartPolicy: OnFailure
      containers:
        - name: jobmanager
          image: hub/online/xxx-flink:flink_on_k8s
          imagePullPolicy: IfNotPresent
          env:
          args: ["standalone-job", "--job-classname", "com.xxx.App"]
          ports:
            - containerPort: 6123
              name: rpc
            - containerPort: 6124
              name: blob-server
            - containerPort: 8081
              name: webui
          livenessProbe:
            tcpSocket:
              port: 6123
            initialDelaySeconds: 30
            periodSeconds: 60
          volumeMounts:
            - name: flink-config-volume
              mountPath: /opt/flink/conf
      volumes:
        - name: flink-config-volume
          configMap:
            name: flink-config
            items:
              - key: flink-conf.yaml
                path: flink-conf.yaml
              - key: log4j-console.properties
                path: log4j-console.properties

flink-taskmanager.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: xxx-flink-taskmanager
spec:
  replicas: 1
  selector:
    matchLabels:
      app: flink
      component: taskmanager
  template:
    metadata:
      labels:
        app: flink
        component: taskmanager
    spec:
      containers:
        - name: taskmanager
          image: hub/online/xxx-flink:flink_on_k8s
          imagePullPolicy: IfNotPresent
          env:
          args: ["taskmanager"]
          ports:
            - containerPort: 6122
              name: rpc
            - containerPort: 6125
              name: query-state
          livenessProbe:
            tcpSocket:
              port: 6122
            initialDelaySeconds: 30
            periodSeconds: 60
          volumeMounts:
            - name: flink-config-volume
              mountPath: /opt/flink/conf/
          securityContext:
            runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessary
      volumes:
        - name: flink-config-volume
          configMap:
            name: flink-config
            items:
              - key: flink-conf.yaml
                path: flink-conf.yaml
              - key: log4j-console.properties
                path: log4j-console.properties

三、定义ConfigMap

通过flink-config-configmap.yaml文件将它们定义为 ConfigMap 来实现配置的传递和读取。如果使用默认配置，这一步则不需要。
flink-config-configmap.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  name: flink-config
  labels:
    app: flink
data:
  flink-conf.yaml: |+
    jobmanager.rpc.address: xxx-flink-jobmanager
    taskmanager.numberOfTaskSlots: 1
    blob.server.port: 6124
    jobmanager.rpc.port: 6123
    taskmanager.rpc.port: 6122
    queryable-state.proxy.ports: 6125
    jobmanager.memory.process.size: 1600m
    taskmanager.memory.process.size: 1728m
    parallelism.default: 1
  log4j-console.properties: |+
    rootLogger.level = INFO
    rootLogger.appenderRef.console.ref = ConsoleAppender
    rootLogger.appenderRef.rolling.ref = RollingFileAppender
    logger.akka.name = akka
    logger.akka.level = INFO
    logger.kafka.name= org.apache.kafka
    logger.kafka.level = INFO
    logger.hadoop.name = org.apache.hadoop
    logger.hadoop.level = INFO
    logger.zookeeper.name = org.apache.zookeeper
    logger.zookeeper.level = INFO
    appender.console.name = ConsoleAppender
    appender.console.type = CONSOLE
    appender.console.layout.type = PatternLayout
    appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
    appender.rolling.name = RollingFileAppender
    appender.rolling.type = RollingFile
    appender.rolling.append = false
    appender.rolling.fileName = ${sys:log.file}
    appender.rolling.filePattern = ${sys:log.file}.%i
    appender.rolling.layout.type = PatternLayout
    appender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
    appender.rolling.policies.type = Policies
    appender.rolling.policies.size.type = SizeBasedTriggeringPolicy
    appender.rolling.policies.size.size=100MB
    appender.rolling.strategy.type = DefaultRolloverStrategy
    appender.rolling.strategy.max = 10
    logger.netty.name = org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline
    logger.netty.level = OFF

四、启动JobManager

JobManager 的执行过程分为两步:

首先，JobManager 通过 Deployment 进行描述，保证 1 个副本的 Container 运行 JobManager，可以定义一个标签，例如 flink-jobmanager。

kubectl apply -f flink-jobmanager.yaml

其次，还需要定义一个JobManager的Service，通过 service name 和 port 暴露 JobManager 服务，通过标签选择对应的 pods。

kubectl apply -f flink-jobmanager-service.yaml

flink-jobmanager-service.yaml

apiVersion: v1
kind: Service
metadata:
  name: flink-jobmanager
spec:
  type: ClusterIP
  ports:
    - name: rpc
      port: 6123
    - name: blob-server
      port: 6124
    - name: webui
      port: 8081
  selector:
    app: flink
    component: jobmanager

五、启动TaskManager

TaskManager 也是通过 Deployment 来进行描述，保证 n 个副本的 Container 运行 TaskManager，同时也需要定义一个标签，例如 flink-taskmanager。

kubectl apply -f flink-taskmanager.yaml

六、更新Jar包操作

更新镜像中的Jar包，重新推镜像，更新完成后，修改flink-jobmanager.yaml和flink-taskmanager.yaml文件中的image路径，然后重新执行

kubectl apply -f flink-jobmanager.yaml
kubectl apply -f flink-taskmanager.yaml