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
# 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

3. 集群知识

k8s上的两种模式:Native和Standalone

Flink Kubernetes Operator 支持:原生部署native(默认)和独立部署standalone

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

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
...
spec:
  ...
  mode: standalone

两种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

注: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

注:为了方便访问,可以再创建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实现,下面仅列出修改部分,注意flinkConfigurationpodTemplate两部分:

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

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

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

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

6. 补充

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

K8S/Kubernetes社区为您提供最前沿的新闻资讯和知识内容

更多推荐