【梳理】k8s使用Operator搭建Flink集群（高可用可选）_flink-kubernetes-operator

1. 架构图

参考：部署验证demo

2. helm 安装operator

安装cert-manager依赖
Jetstack/cert-manager 是 Kubernetes 生态系统中的一款开源项目，它提供了一种自动化的方式来管理 TLS 证书的生命周期

kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml
1

# helm 安装 , 包含 deploy*1 、cm*1、crd*2 以及 rbac sa webhook
kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml
helm repo add flink-operator-repo https://downloads.apache.org/flink/flink-kubernetes-operator-1.7.0/
helm install flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator
1
2
3
4

3. 集群知识

k8s上的两种模式：Native和Standalone

Flink Kubernetes Operator 支持：原生部署native（默认）和独立部署standalone

可以使用部署规范中的 mode 字段设置部署模式。

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
...
spec:
  ...
  mode: standalone
1
2
3
4
5
6

两种CR

FlinkDeployment CR 定义了 Flink 应用程序和会话集群部署。

FlinkSessionJob CR 定义了 Session 集群上的会话任务（Job）

Flink Kubernetes Operator 支持两种主要类型的部署：Application集群 和 Session集群 ，k8s上不支持Job集群。

4. 运行集群实例

Demo1：Application 集群

一次性任务，只有一个Job，执行器和逻辑代码打包成一个jar，直接运行，运行即结束。

# 此crd创建后，operator会创建：
# 1个deploy（即jobmanager,镜像为flink:1.17）
# 1个pod（即taskManager,镜像也是flink:1.17,任务jar包在镜像中）
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: basic-example
spec:
  image: flink:1.17
  flinkVersion: v1_17
  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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

注：k8s不支持Job集群，可以看做是 Flink Application 集群”客户端运行“的替代方案。集群管理器为每个提交的作业启动一个集群。

Demo2：Session集群

多租户，多个job，每个sessionjob代表一个job，有提交jar包的功能。
Session使用与Application 集群类似的规范，唯一的区别是 job 未定义。

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: basic-session-deployment-example
spec:
  image: flink:1.17
  flinkVersion: v1_17
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  serviceAccount: flink
---
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
  name: basic-session-job-example
spec:
  deploymentName: basic-session-deployment-example
  job:
    jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.16.1/flink-examples-streaming_2.12-1.16.1-TopSpeedWindowing.jar
    parallelism: 4
    upgradeMode: stateless

---
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
  name: basic-session-job-example2
spec:
  deploymentName: basic-session-deployment-example
  job:
    jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.16.1/flink-examples-streaming_2.12-1.16.1.jar
    parallelism: 2
    upgradeMode: stateless
    entryClass: org.apache.flink.streaming.examples.statemachine.StateMachineExample
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

注：为了方便访问，可以再创建ingress。svc端口默认为8081，指向svc-rest即可。

优劣

• Session集群，是常规的共享方式。
• Application集群，提供了更好的隔离，生命周期与程序逻辑有关。
• Job集群，适合长期运行、要求高稳定性的大型作业。（启动慢）

5. 高可用部署

Flink 提供了两种高可用服务实现：

• ZooKeeper：每个 Flink 集群部署都可以使用 ZooKeeper HA 服务。它们需要一个运行的 ZooKeeper 复制组（quorum）。
• Kubernetes：Kubernetes HA 服务只能运行在 Kubernetes 上。
  注：Flink 持久化元数据和 job 组件，直到作业执行成功、被取消或最终失败，再删除。

我在实践时，遇到两个问题：

问题1：High availability should be enabled when starting standby JobManagers

直接增加jobManager的副本数提示上述错误，要先做选举配置，我选择的是k8s实现，下面仅列出修改部分，注意flinkConfiguration和podTemplate两部分：

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
spec:
  flinkConfiguration:
    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
    high-availability.storageDir: file:///flink-data/ha
    web.upload.dir: /flink-data  #会自动创建flink-web-upload目录保存上传的jar包
  jobManager:
    replicas: 2
  taskManager:
    replicas: 2
  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          volumeMounts:
            - mountPath: /flink-data
              name: flink-volume
      volumes:
        - name: flink-volume
          persistentVolumeClaim:
            claimName: flink-ha
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

问题2：The base directory of the JobResultStore isn’t accessible

如果没有正确挂载存储卷提示上述错误。
官方demo配置的卷是宿主机的路径，如下：

      volumes:
        - name: flink-volume
          hostPath:
            path: /tmp/flink  # 如果宿主机上没有这个路径会报错
            type: Director
1
2
3
4
5

6. 补充

1. 拉取国外镜像比较困难，可以使用https://dockerproxy.com/
2. 搭建多租户PaaS平台，可以使用Session方式，新任务可以通过CR进行管理，也可以由页面添加jar包。
3. 本文内容来源于Flink官网，进行翻译、简化、整理，供大家参考~