如果你正在做实时计算迟早要面对一个问题Flink 集群怎么在 Kubernetes 上稳定地跑起来。自己手写 YAML 一条条部署当然也能跑但当作业数量上来之后升级、扩缩容、故障恢复全都变成手工活稍不注意就把线上搞挂。flink-operator 的出现就是用 Kubernetes 声明式的方式统一管理 Flink 集群你只需要提交一个 FlinkDeployment 对象创建、更新、销毁一套 Flink 集群的活都交给 Operator 控制器去处理。这篇文章面向正在调研或准备落地的开发和运维同学讲清楚 operator 的安装、第一个 Flink 集群的部署、核心参数怎么配以及一堆实际部署中才会遇到的坑。1. 为什么选择 flink-operator 而不是自己写 YAML1.1 手动部署 Flink 到 Kubernetes 的真实痛点先说说我自己最早是怎么干的。一个 Flink 集群在 Kubernetes 上跑起来表面上只需要拉起 JobManager 和 TaskManager 两组 Pod但真正要交付生产环境时需要准备的东西远不止两个 Deployment。JobManager 要暴露 REST 和 Web UI 端口TaskManager 之间要互相通信这意味着至少三个 ServiceFlink 的配置文件、日志配置、高可用元数据存储通常要拆成 ConfigMap、PVC 等对象如果你要把 Dashboard 暴露给外部访问还得再补 Ingress。一个最小可用的集群YAML 加在一起轻松超过两百行。这还只是初始创建真正痛苦的是后面的变更。升级 Flink 版本的时候你要改镜像 tag还要担心 JobManager 和 TaskManager 镜像不一致导致的兼容问题调整并行度的时候你得手动算 TaskManager 副本数改完 Deployment 的 replicas 之后还要盯着滚动更新JobManager 单点挂掉的时候Kubernetes 只会帮你重新拉起一个 Pod但状态、Checkpoint 元数据、作业恢复这些事它一概不管。做过一轮之后你就会意识到这种方式适合尝鲜不适合长期运维。1.2 Operator 模式带来的变化Operator 模式本质上解决的是“应用生命周期管理自动化”的问题。它把运维人员对 Flink 集群的认知固化到一段控制逻辑里通过自定义资源CRD暴露给上层用户。你只需要提交一个 FlinkDeployment 对象声明“我要一个 Flink 1.17 的集群2 个 TaskManager slot作业并行度 4”Operator 会负责生成底层所有 Kubernetes 资源并持续保证实际状态向期望状态收敛。用一个生活化的类比传统手动部署就像你每次开店都要自己进货、摆货架、算库存而 Operator 模式就像一个店长你只需要告诉它“我要开一家什么样的店”它自己去搞定货架、补货、盘点甚至打折促销。省下来的时间你可以真正去关心业务代码和作业稳定性。在实际项目中我最大的体感是以前需要写脚本去完成的发布动作现在变成了一条kubectl apply -f xxx.yaml以前升级需要人工编排的“先起新 JobManager、再做 Savepoint、再切换流量”流程变成了 Operator 内部的一个内置步骤。它更适合多团队、多作业、需要频繁发布和回滚的中大型实时计算场景。2. flink-kubernetes-operator 的原理与版本选型2.1 一次调谐流程是怎么跑起来的flink-kubernetes-operator 安装到集群后核心是一组控制器 Pod它们监听 FlinkDeployment 资源的变化。当用户执行kubectl apply提交一个 FlinkDeployment 时请求会先经过准入 Webhook 做参数校验和默认值填充然后写入 etcd。控制器通过 List-Watch 机制拿到这个新对象进入调谐循环。它把期望状态CR 里的 spec和当前状态集群中已存在的 Deployment、Service、ConfigMap做对比然后执行一系列操作创建 JobManager Deployment、创建 REST Service、计算 TaskManager 副本数并创建对应 Deployment、写入 Flink 配置文件、配置 Ingress 等。操作完成后控制器把结果写回 FlinkDeployment 的 status 字段。所以你会看到 FlinkDeployment 对象上有类似jobManagerDeploymentStatus: READY、jobStatus.state: RUNNING这样的状态信息。这些状态不是人工维护的而是控制器每次调谐后自动更新的。理解了这一点排障的时候就会很自然地去看 status 和 events而不是一头扎进 YAML 里瞎改。2.2 CRD 版本选择FlinkCluster 还是 FlinkDeployment这是很多人第一次接触时会踩的坑。早期版本的 flink-kubernetes-operator 使用FlinkCluster作为核心 CRDAPI 组是flink.apache.org/v1alpha1。后来社区为了更好地区分“集群管理”和“作业管理”引入了FlinkDeploymentAPI 版本升级为v1beta1把集群和应用的生命周期统一到一个对象里。如果你是从网上翻到的一些老教程里面很可能是FlinkCluster的写法直接套用到新版 Operator 上通常会报no matches for kind FlinkCluster。我建议你以官方文档当前版本为准新项目一律使用FlinkDeployment。它不仅能描述 JobManager 和 TaskManager 的拓扑还能描述作业的 jar 包、并行度、升级模式一个 CR 覆盖所有诉求。如果你在集群里已经存在旧的 FlinkCluster 资源升级 Operator 之前需要先确认兼容性。社区通常会在 release notes 里说明哪些旧 API 仍然可用但长期维护的角度看还是尽早迁移到 FlinkDeployment 更省心。2.3 版本组合怎么选Operator、Flink、Kubernetes 三者之间存在版本匹配关系不能随意组合。根据我目前的经验以下几个组合比较稳组件建议版本说明Kubernetes1.25 及以上1.27 / 1.28 实测正常旧版本需要确认 CRD 兼容性Helm3.8 及以上安装和升级 Operator 的主要方式flink-kubernetes-operator1.7.x 系列API 稳定支持 Flink 1.17 / 1.18Flink1.17 或 1.18与 Operator 1.7.x 配合成熟社区支持度好共享存储支持 ReadWriteMany生产环境做高可用和 Checkpoint 的硬性要求选版本时最怕的是“看着差不多就上了”。比如拿一个较新的 Operator 去管理旧版 Flink 集群API 里新增的字段会被忽略但某些默认行为可能直接导致集群起不来。我自己的习惯是先到官方文档查看版本兼容矩阵再在小集群上做一次冒烟测试确认没问题后再推到生产。3. 环境准备与 Operator 安装3.1 动手前先过一遍环境清单安装 Operator 之前先把基础环境确认好。Kubernetes 集群要保持可用状态kubectl能正确连接集群Helm 版本不要太老我建议 3.8 以上集群内部需要允许创建和更新 CRD 资源也就是你要具备足够的 RBAC 权限。另外要确认集群是否启用了 admission webhook 支持。flink-kubernetes-operator 安装时会注册一个 ValidatingWebhookConfiguration 用于校验 FlinkDeployment 参数。如果你的集群环境比较特殊比如某些受管集群默认禁用了 webhook 扩展能力安装后 Operator 可能无法正常工作。遇到这种情况需要先查询集群是否支持admissionregistration.k8s.io/v1这个 API 组。存储方面如果只是本地测试可以暂时不准备共享存储但如果你打算跑生产集群尤其是要开启高可用模式一定要提前准备一个 ReadWriteMany 的 StorageClass。这个点后面讲 HA 时会详细说现在先记在心里。3.2 Helm 方式安装 Operator安装 flink-kubernetes-operator 最主流的方式是 Helm。把官方 Helm 仓库添加到本地后直接安装即可。由于仓库地址在不同版本中可能会调整我建议你在动手前到项目官方文档里复制最新的仓库地址别凭记忆拼写。helm repo add flink-operator FLINK_OPERATOR_REPO_URL helm repo update然后创建命名空间并安装kubectl create namespace flink helm install flink-kubernetes-operator flink-operator/flink-kubernetes-operator \ --namespace flink \ --create-namespace安装过程会创建 FlinkDeployment 相关的 CRD、服务账号、RBAC 权限、Operator Deployment 和 Webhook 配置。默认情况下Operator 会监听所有命名空间中的 FlinkDeployment 资源如果你希望限制监听范围可以设置watchNamespaces参数。比如只允许在default和flink-demo两个命名空间里创建 Flink 集群helm install flink-kubernetes-operator flink-operator/flink-kubernetes-operator \ --namespace flink \ --set watchNamespaces{default,flink-demo}这个参数在生产环境很有用可以避免 Operator 的服务账号权限被滥用。但要注意设置之后其他命名空间里的 FlinkDeployment 就不会被处理如果你后续有新增命名空间的需求需要重新升级 Helm release。3.3 验证安装是否正常安装完成后先确认 Pod 状态和 CRD 是否注册成功。kubectl -n flink get pods kubectl get crd | grep flink.apache.org正常情况下Opeartor 的 Pod 应该处于 Running 状态CRD 列表里能看到flinkdeployments.flink.apache.org等条目。接着查看 Operator 日志确认没有证书相关的报错kubectl -n flink logs deployment/flink-kubernetes-operator这里我想提醒一句Operator 的启动日志里如果出现了 x509 证书过期或 webhook 注册失败的提示多半是 Helm chart 自带的证书生成逻辑出了问题。比较常见的解决办法是删除相关 Secret 后重新执行helm upgrade让 chart 重新生成证书。如果集群里已经安装了 cert-manager也可以配置 chart 使用 cert-manager 管理证书但这需要额外修改 values 文件建议在正式使用前多做一次验证。4. 部署第一个 Flink 集群4.1 准备命名空间和服务账号Operator 安装好之后先创建一个专门跑 Flink 集群的命名空间并准备一个服务账号。FlinkDeployment 里可以指定serviceAccount这个服务账号会被 JobManager 和 TaskManager 使用。如果你的作业需要访问 Kubernetes API比如使用 Kubernetes 原生 HA这个服务账号还需要绑定相应 RBAC 权限。kubectl create namespace flink-demo kubectl -n flink-demo create serviceaccount flink如果暂时不需要作业访问集群内的其他资源使用默认服务账号也可以但我还是建议显式创建一个专用服务账号方便后续做权限隔离和审计。4.2 写一个最小 FlinkDeployment创建一个basic-example.yaml内容如下apiVersion: flink.apache.org/v1beta1 kind: FlinkDeployment metadata: name: basic-example namespace: flink-demo spec: image: flink:1.17.2 flinkVersion: v1_17 flinkConfiguration: taskmanager.numberOfTaskSlots: 2 serviceAccount: flink jobManager: replicas: 1 resource: memory: 1024m cpu: 1 taskManager: resource: memory: 1024m cpu: 1注意几个关键点。image使用的是官方 Flink 镜像flinkVersion的格式是v1_17下划线而不是点。这个字段必须和镜像里的 Flink 大版本保持一致否则 Operator 在生成配置时可能计算出错误的内存参数。taskmanager.numberOfTaskSlots表示每个 TaskManager 提供的 slot 数量这里设置了 2意味着后续作业如果并行度是 4Operator 会拉起 2 个 TaskManager。4.3 查看状态与访问 Dashboard用kubectl apply提交 CR 后观察资源创建情况kubectl apply -f basic-example.yaml kubectl -n flink-demo get flinkdeployments kubectl -n flink-demo get pods第一次创建时JobManager 和 TaskManager 的 Pod 会依次启动。你可以用kubectl describe flinkdeployment basic-example查看详细状态status 里如果出现jobManagerDeploymentStatus: READY说明集群已经正常拉起。访问 Web UI 最简单的方式是端口转发kubectl -n flink-demo port-forward svc/basic-example-rest 8081:8081浏览器打开localhost:8081就能看到 Flink Dashboard。这里要注意 Service 的名称规则如果 FlinkDeployment 叫basic-example对应的 REST Service 一般是basic-example-rest不是basic-example。我见过不少人在这里卡住反复检查 Service 配置其实只是名字没找对。4.4 用示例作业跑通全链路集群起来了接着验证作业提交能力。官方 Flink 镜像自带了示例 jar可以在 FlinkDeployment 里直接声明要运行的作业。修改 YAML加上job部分spec: job: jarURI: local:///opt/flink/examples/streaming/WindowWordCount.jar parallelism: 2 upgradeMode: savepoint state: running重新 apply 之后Operator 会在集群启动完成后自动提交该作业。查看作业状态kubectl -n flink-demo get flinkdeployments basic-example -o jsonpath{.status.jobStatus.state}如果输出是RUNNING说明从集群创建到作业提交的全链路已经跑通。upgradeMode: savepoint这个配置很关键它决定了后续你更新作业或升级版本时Operator 是否会先做 Savepoint 再优雅重启。对于生产环境我强烈建议配置成savepoint不要图省事用stateless。5. 核心配置深度解析资源、高可用与监控5.1 内存和 CPU 配置的正确姿势资源配额是部署 FlinkDeployment 时最容易出问题的地方。很多人的第一反应是给 Pod 设置一个很大的内存然后发现 JobManager 反复重启。原因很简单Flink 的 JVM 内存分为堆内、堆外、JVM Overhead、网络缓冲区等多个部分仅设置 Pod 总内存并不能决定各个部分如何分配。Operator 在生成 Flink 配置时会根据jobManager.resource.memory和taskManager.resource.memory自动推导 JVM 参数。如果你同时在flinkConfiguration里手动写了jobmanager.memory.process.size、taskmanager.memory.process.size等参数就很可能和 Pod 资源产生冲突。我的经验是二选一。要么只设置 Pod 的resource.memory和resource.cpu让 Operator 和官方镜像的 entrypoint 脚本自动计算要么你把flinkConfiguration里的内存参数填得明明白白同时把 Pod 内存也对应调整。内存大小的选择也没有绝对标准。JobManager 主要承担调度和协调工作一般 1-2G 足够TaskManager 是真正干活的地方建议至少给到 2-4Gslot 数量根据自己的作业并发情况设置。注意slot 数量并不是越多越好一个 slot 对应一个执行线程如果 CPU 和内存不足slot 再多也只是把线程切换开销变大。5.2 高可用从单副本到多副本默认情况下jobManager.replicas是 1也就是只有一个 JobManager。这个配置在测试环境没问题生产环境却是绝对不可接受的。JobManager 一旦异常宕机集群会失去调度能力所有运行中的作业都会失败。Flink 在 Kubernetes 上支持原生 HA 模式原理是利用 Kubernetes 的 ConfigMap 和 Leader Election 机制让多个 JobManager 实例选举出一个 Leader其余实例作为 Standby。配置方式如下spec: jobManager: replicas: 2 flinkConfiguration: high-availability: kubernetes high-availability.storageDir: 共享存储路径high-availability.storageDir是 HA 元数据的存放位置JobManager 会把作业图、Checkpoint 元数据等信息写到这里。Standby JobManager 启动时会从这个目录恢复状态接管后无缝继续运行。这里有个非常容易踩的坑storageDir必须是所有 JobManager 副本都能访问的共享存储而且存储必须支持多节点同时挂载。如果你选了普通的单节点读写 PVC第二个 JobManager 在另一台机器上启动时根本读不到数据。生产环境请务必准备 ReadWriteMany 的存储方案比如共享文件系统或对象存储。5.3 监控与日志的接入思路Flink 自身暴露了大量的 Metrics包括吞吐量、延迟、Checkpoint 耗时、背压情况等。要采集这些数据最常用的方案是 Prometheus。在flinkConfiguration里配置 Prometheus Reporterspec: flinkConfiguration: metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory metrics.reporter.prom.port: 9249然后通过 Prometheus 的 ServiceMonitor 或手工配置抓取任务来收集指标。如果你还没有 Prometheus也可以先在本地端口转发 9249 端口看一眼指标是否正常输出。日志方面Flink 默认会输出到 stdout容器运行时可以收集这些日志。如果你需要更精细的日志配置可以通过 FlinkDeployment 的spec.logConfig字段覆盖默认的 log4j 配置。实际项目中我通常是把日志输出到 stdout再由日志采集组件统一收集和检索这样省心且不会丢日志。6. 日常运维升级、扩缩容与资源回收6.1 Flink 版本和作业升级使用 Operator 之后升级 Flink 版本变成了一件“改一行配置”的事。你只需要修改 FlinkDeployment 中的image为新的 Flink 镜像版本然后kubectl apply。Operator 检测到字段变化后会执行滚动升级先处理 JobManager再根据计算结果调整 TaskManager。如果作业配置了upgradeMode: savepoint升级过程中 Operator 会先让作业停止并生成 Savepoint然后用新镜像重新启动作业并从该 Savepoint 恢复状态。这个过程通常不用人工干预是 Operator 最有价值的能力之一。但升级过程中有一个细节需要注意新版本镜像里必须存在jarURI指向的文件。如果你把作业 jar 放在自定义镜像里升级前一定要确认新镜像已经包含新的 jar。否则作业会因为没有提交物而启动失败然后一直处于 RESTARTING 状态。我踩过一次这个坑排查了半天才发现是 jar 路径在新镜像里不存在。6.2 并行度与 TaskManager 数量调整很多人会误以为 TaskManager 数量是手动设置的其实在 Operator 模式下TaskManager 的数量是根据作业并行度和每个 TaskManager 的 slot 数自动计算的。并行度 4、每个 TaskManager 2 个 slot就会自动拉起 2 个 TaskManager。这个设计让用户只需要关心“作业要多少并行度”而不需要关心底层 Pod 数量。当你调整spec.job.parallelism时Operator 会触发一次带 Savepoint 的作业重启并根据新的并行度和 slot 数自动调整 TaskManager 数量。如果你只调整taskmanager.numberOfTaskSlots同样会触发资源拓扑变化。生产环境中调整并行度前最好先评估一下状态大小和状态后端因为并行度变化可能会影响状态分布Savepoint 恢复速度慢会直接影响作业可用性。6.3 删除集群后如何处理残留资源删除 FlinkDeployment 时Operator 会清理对应的 Deployment、Service、ConfigMap 等资源但 PVC 和 Savepoint 数据可能不会自动删除这是很多运维同学容易忽略的地方。如果你创建集群时配置了 PVC 用于存储删除 CR 后 PVC 依然存在。这是有意设计的为了防止误删数据。想彻底清理时需要手动删除 PVCkubectl -n flink-demo get pvc kubectl -n flink-demo delete pvc PVC名称同时如果你用了对象存储保存 Savepoint那部分数据也需要根据清理策略单独处置。我的建议是在删除测试集群前先确认没有需要保留的作业状态再一次性清理 PVC 和存储目录。否则重建同名集群时可能会因为旧 PVC 被占用或元数据残留导致新集群起不来。7. 常见问题排查速查表与实战心得7.1 高频问题排查速查表部署和维护过程中我积累了一些高频问题的排查经验整理成表格供你参考现象常见原因排查思路Operator Pod CrashLoopBackOffWebhook 证书生成异常或过期查看 Operator 日志删除证书 Secret 后重新执行helm upgradeFlinkDeployment 创建后一直 Pending资源配额不足、StorageClass 不存在kubectl describe pod查看 Events确认 PVC 和节点资源JobManager 反复 CrashLoopBackOff内存配置自相矛盾删掉flinkConfiguration里手动设置的 memory 参数只保留 Pod 资源作业一直 RESTARTING用户代码异常、Checkpoint 存储不可用查看 JobManager 日志检查 Savepoint 和 Checkpoint 路径的连通性Dashboard 无法访问Service 名称不对、端口转发目标错误用kubectl get svc确认 Service 名称REST 服务通常以-rest结尾这套速查表覆盖了大部分初级问题。归根结底排障的顺序一定是先看事件再看日志最后才去怀疑 Operator 本身。大多数问题并不是 Operator 的 bug而是配置不当或环境不满足预期。7.2 两个容易被忽略的坑第一个坑是证书过期。flink-kubernetes-operator 安装时生成的 Webhook 证书通常有有效期限制。证书过期后你再向集群提交 FlinkDeployment 时请求会被 Webhook 拦截报出类似failed calling webhook的错误。这时候最直接的处理方法是重新执行一次helm upgrade让证书重新生成或者按官方文档配置证书自动续期。第二个坑是删除集群后立即重建同名集群。由于旧集群的 PVC 和服务资源可能还处于 Terminating 状态新集群如果同名控制器在创建时可能因为资源冲突而一直等待。最简单的方法是把kubectl delete flinkdeployment等你确定资源完全清理后再重新创建。7.3 最后的实操心得我个人在实际操作中有几条心得。第一版本一定要成套升级不要只升级 Operator 或只升级 Flink两者对不上是最难排查的隐形问题。第二任何新方案先在一个小集群上跑通全流程再引入生产环境。我见过不少团队在 UAT 环境验证都没问题一上生产就翻车差距往往就在存储、RBAC 这类容易被忽视的基建上。第三生产环境必须开启 HA哪怕业务规模再小也不要用单副本 JobManager 赌稳定性。这四个问题处理完之后flink on k8s 这条路基本就比较顺畅了。