企业官网建设流程全解析

1. 这不是“装个软件”那么简单：Flink集群搭建的真实定位与价值锚点

很多人看到“Flink集群搭建”这六个字，第一反应是——不就是下载几个包、改几行配置、起几个进程吗？我当年也是这么想的，直到在某次实时风控项目压测中，集群在QPS刚过8000时突然出现TaskManager频繁心跳超时、背压指标像心电图一样乱跳、Checkpoint持续失败……最后排查下来，根本不是代码逻辑问题，而是最初搭建时图省事用了单机伪分布式模式硬扛生产流量，连最基本的资源隔离和网络拓扑都没做。Flink从来就不是个“开箱即用”的玩具框架，它的集群形态直接决定了你后续能跑多稳、跑多快、跑多远。

所以，我们今天聊的“快速入门Flink (2) —— Flink 集群搭建”，核心关键词其实是稳定性边界、资源可伸缩性、故障恢复能力这三个硬指标。它解决的不是“能不能跑起来”，而是“能不能在业务峰值下不掉链子”、“当一个节点宕机时任务是否自动漂移”、“Checkpoint失败后能否从最近一致点无缝续跑”。适合谁？不是刚学完WordCount的新手，而是已经写过至少3个真实流处理作业、正准备把本地测试环境迁移到团队共享集群的中级开发者；或者是运维同学，被开发拉着一起看YARN日志、查JVM GC停顿、调Network Buffers大小的那类人。如果你还在纠结“Flink和Spark Streaming到底选哪个”，那建议先回去把Flink官网的Architecture Overview读三遍——集群不是容器，它是整个实时计算体系的承重墙。

我实测过三种典型场景下的集群表现差异：单机Standalone（仅用于本地调试）、3节点YARN Session（中小规模数据中台）、5节点Kubernetes Native（高可用金融级实时数仓）。它们的启动命令看起来差不多，但背后涉及的线程模型、内存分片策略、RPC超时机制、状态后端选型，全都不一样。比如同样是设置taskmanager.memory.process.size=4g，在Standalone下它真就占4G，在YARN里可能被YARN Container Memory Overhead吃掉1.2G，剩下不到3G给JVM堆和堆外内存分，结果Flink自己还没启动，YARN就给你Kill了。这些坑，光看文档是填不完的，得靠一次一次搭、一次一次崩、一次一次翻源码里的MemorySize类注释才能真正理解。

2. 搭建前必须厘清的四大底层逻辑：为什么不能照着教程抄配置？

2.1 Flink集群不是“服务集合”，而是一个协同演化的状态机系统

很多新手误以为Flink集群 = JobManager + TaskManager + ZooKeeper（或HA存储），只要三个进程都起来了，就算搭好了。这是最危险的认知偏差。Flink真正的集群状态，是由JobManager的高可用协调器（LeaderElectionService）、TaskManager的心跳注册中心（ResourceManager）、每个TaskManager内部的Slot生命周期管理器（SlotPool）、以及外部状态后端（StateBackend）与检查点存储（CheckpointStorage）四者共同维护的一个动态闭环。任何一个环节的配置失配，都会导致状态不一致。

举个具体例子：你在flink-conf.yaml里设置了high-availability: zookeeper，但没配high-availability.zookeeper.quorum，JobManager启动时会报错退出；可如果你配了ZooKeeper地址，却忘了在ZK里手动创建/flink根路径（Flink不会自动创建），JobManager会静默卡在Leader选举阶段，日志里只有一行Waiting for leader to be elected...，没有任何错误提示。这种“半死不活”的状态，比直接报错更难排查。原因在于Flink的HA模块设计哲学是“宁可不启动，也不启动一个不可靠的Leader”，它把一致性放在了可用性之前。

再比如state.backend: rocksdb这个配置，表面看只是换了个状态后端，实际它强制绑定了state.backend.rocksdb.predefined-options: DEFAULT，而DEFAULT预设值在不同Flink版本里差异极大：1.13默认用OptimizedForSmallData，1.15改成OptimizedForLargeData，前者对小状态友好但GC压力大，后者对大状态吞吐高但初始化慢。如果你从1.13升级到1.15，没同步调整RocksDB参数，就会发现Checkpoint时间从2秒飙升到15秒——这不是集群没搭好，而是状态后端的隐式契约被打破了。

2.2 网络通信不是“通就行”，而是要穿透三层隔离模型

Flink集群的网络通信，必须同时满足进程间通信（IPC）、跨节点数据传输（Shuffle）、外部系统对接（Kafka/HDFS）三重需求。这导致它天然面临三层隔离：

• OS层隔离：Linux内核的net.core.somaxconn（连接队列长度）、net.ipv4.tcp_tw_reuse（TIME_WAIT复用）、vm.swappiness（交换分区倾向）直接影响TaskManager之间Shuffle数据的吞吐；
• JVM层隔离：Flink的Netty Server默认使用EpollEventLoopGroup（Linux特有），如果部署在Docker容器里没加--privileged或没挂载/dev/shm，会自动fallback到NioEventLoopGroup，性能下降30%以上；
• Flink层隔离：taskmanager.network.memory.fraction这个参数，控制的是TaskManager总内存中划给网络缓冲区的比例，默认0.1，但如果你的作业大量使用rebalance()或shuffle()算子，这个值必须调到0.2甚至0.3，否则Buffer Pool耗尽会导致反压直接打满上游。

我遇到过最典型的案例：某公司用Kubernetes部署Flink，所有Pod都跑了，kubectl get pods全是Running，但作业一提交就卡在SCHEDULED状态。最后发现是K8s Service的ClusterIP模式下，JobManager通过jobmanager.rpc.address解析到的是ClusterIP，而TaskManager注册时上报的是Pod IP，两者网络不通。解决方案不是改DNS，而是必须显式配置jobmanager.rpc.bind-address: 0.0.0.0并配合rest.bind-address: 0.0.0.0，让JobManager监听所有接口，再通过K8s Headless Service做DNS SRV记录发现。这个细节，90%的入门教程都不会提，因为它已经超出“搭建”范畴，进入“云原生网络治理”领域了。

2.3 内存模型不是“堆内存+非堆”，而是五层嵌套的精密仪表盘

Flink的内存模型是业界最复杂的之一，它把JVM内存拆成了5个独立可调区域：

关键陷阱在于：这五层不是简单相加，而是存在强依赖关系。比如taskmanager.memory.process.size=4g是总进程内存上限，它等于JVM Heap + Off-heap Direct + Network Memory + Managed Memory + Framework Heap + JVM Metaspace + JVM Overhead之和。其中JVM Overhead是Flink自动计算的（默认为process.size * 0.1），Metaspace是JVM自己管的（默认256M）。如果你手动设了jvm.heap=2g、direct=1g、network.fraction=0.2（即0.8g）、managed.fraction=0.4（即1.6g），加起来已经4.4g，超出了process.size，Flink启动时会直接拒绝，报错Total memory configuration exceeds configured process memory size。

更隐蔽的是Managed Memory的分配时机：它只在TaskManager启动时静态分配，之后无法动态扩容。如果你的作业里有个CoProcessFunction需要大量状态缓存，而Managed Memory又不够，RocksDB就会被迫写磁盘，I/O飙升。这时候调大managed.fraction也没用——因为进程总内存已锁死。唯一解法是重启TaskManager，并确保process.size留足余量。这就是为什么我坚持在生产环境永远按公式计算：process.size = max(4g, 1.5 × (heap + direct + network + managed))，宁可浪费1G内存，也不能让Managed Memory成为瓶颈。

2.4 高可用（HA）不是“加个ZooKeeper”，而是定义故障恢复SLA

Flink的HA配置，本质是在回答一个问题：“当JobManager崩溃时，你能接受多长时间的服务中断？”答案直接决定你的HA架构选型：

• 无HA模式（high-availability: NONE）：中断时间 = JobManager进程重启时间 + 所有TaskManager重新注册时间 + 最新Checkpoint恢复时间。实测平均12~18秒，适用于离线ETL等容忍分钟级延迟的场景；
• ZooKeeper HA（high-availability: zookeeper）：中断时间 = ZK Leader选举时间（通常<2秒）+ 新JobManager从ZK加载JobGraph时间 + TaskManager重连时间。实测平均3~5秒，适合大多数实时大屏、风控预警；
• Kubernetes Native HA（high-availability: kubernetes）：中断时间 = K8s Pod重建时间（含镜像拉取，约8~15秒）+ Flink JobManager StatefulSet主从切换时间。看似比ZK慢，但胜在无需维护ZK集群，且能利用K8s的Liveness Probe自动驱逐异常Pod。

但所有HA模式都有一个致命前提：Checkpoint必须成功且持久化到高可用存储（如HDFS/S3）。如果state.checkpoints.dir指向的是本地磁盘，哪怕ZK选举再快，新JobManager也找不到之前的Checkpoint，只能从Source最新offset重放——这就不是HA，是“灾难恢复”。我见过最惨的案例：某团队把Checkpoint存到TaskManager所在节点的/tmp目录，结果一次磁盘满导致所有TaskManager OOM，ZK里残留的Checkpoint元数据指向的却是已删除的文件，整个集群重启后所有作业从3天前的Kafka offset开始重放……

所以HA配置的核心，不是high-availability.zookeeper.quorum这一行，而是state.checkpoints.dir和state.savepoints.dir这两个路径的存储可靠性。它们必须满足：1）多副本（HDFS replication=3或S3 versioning开启）；2）低延迟（避免用NFS挂载）；3）权限可控（Flink用户必须有读写执行权限）。少一个条件，HA就是纸糊的。

3. 三套生产级集群方案实操：从Standalone到Kubernetes的渐进式落地

3.1 方案一：Standalone集群（仅限开发/测试环境）

Standalone模式最大的价值，不是“轻量”，而是完全暴露Flink原生行为。它不依赖YARN/K8s的资源调度，所有资源分配逻辑都在Flink内部完成，是理解Flink内存、网络、状态机制的“裸金属”沙盒。

部署步骤（以Flink 1.17.1为例）：

1. 下载二进制包并解压：

wget https://archive.apache.org/dist/flink/flink-1.17.1/flink-1.17.1-bin-scala_2.12.tgz tar -xzf flink-1.17.1-bin-scala_2.12.tgz cd flink-1.17.1

2. 修改conf/flink-conf.yaml核心配置（重点标出易错项）：

# 必须显式绑定IP，不能用localhost（否则TaskManager连不上） jobmanager.rpc.address: 192.168.1.100 jobmanager.rpc.port: 6123 jobmanager.bind-host: 0.0.0.0 jobmanager.memory.process.size: 2g # 总进程内存，不是JVM堆！ # TaskManager资源配置（3节点集群示例） taskmanager.numberOfTaskSlots: 4 taskmanager.memory.process.size: 4g # 每个TM独占4G，不是共享！ # 网络缓冲区必须按比例分配，不能写死字节数 taskmanager.network.memory.fraction: 0.2 taskmanager.network.memory.min: 64mb taskmanager.network.memory.max: 2g # 状态后端：开发环境用FsStateBackend够用，但必须指向共享存储 state.backend: filesystem state.checkpoints.dir: file:///data/flink/checkpoints state.savepoints.dir: file:///data/flink/savepoints # 关键！开发环境也要开HA，避免单点故障 high-availability: zookeeper high-availability.storageDir: hdfs://namenode:8020/flink/ha/ high-availability.zookeeper.quorum: zk1:2181,zk2:2181,zk3:2181 high-availability.zookeeper.path.root: /flink

提示：state.checkpoints.dir如果用file://协议，必须确保所有节点的/data/flink/checkpoints是同一块NFS或GlusterFS卷，否则TaskManager A写的Checkpoint，JobManager B读不到。这是Standalone模式下最容易被忽略的分布式一致性陷阱。

3. 启动集群（注意顺序）：

# 先启JobManager（主节点） ./bin/start-cluster.sh # 再启TaskManager（其他节点，需提前配置好SSH免密） ssh node2 "./flink-1.17.1/bin/taskmanager.sh start" ssh node3 "./flink-1.17.1/bin/taskmanager.sh start"

4. 验证集群健康状态：

# 查看JobManager日志，确认"Starting RPC endpoint"和"Starting web frontend" tail -f log/flink-*-jobmanager-*.out # 查看TaskManager日志，确认"Registered at JobManager"和"Number of registered task managers: 2" tail -f log/flink-*-taskmanager-*.out # 用REST API检查Slot可用性（返回JSON中numSlotsAvailable > 0） curl http://192.168.1.100:8081/taskmanagers

实操心得：

• 不要用start-cluster.sh一键启所有节点，它默认只启本机的JM+TM，跨节点必须手动SSH执行；
• taskmanager.sh start命令会后台运行，但不会写pid文件，杀进程得用pkill -f "TaskManagerRunner"；
• Web UI端口（8081）默认只监听localhost，必须加rest.bind-address: 0.0.0.0才对外可见；
• 开发阶段务必打开rest.flamegraph.enabled: true，它能在Web UI里生成CPU火焰图，对排查反压根源极有用。

3.2 方案二：YARN Session集群（中大型企业主流选择）

YARN Session模式的价值，在于复用现有Hadoop生态。它把Flink当作YARN上的一个“长周期应用”，由YARN统一调度资源，Flink只专注计算逻辑。相比Standalone，它天然支持资源弹性伸缩（动态增减TM）、多租户隔离（不同队列）、统一认证（Kerberos）。

部署前提检查清单：

• ✅ Hadoop 3.1+（YARN ResourceManager必须开启yarn.resourcemanager.scheduler.class=org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler）
• ✅ HDFS高可用（NameNode HA + JournalNode）
• ✅ Flink二进制包已上传至HDFS（hdfs dfs -put flink-1.17.1.tar.gz /apps/flink/）
• ✅ 客户端机器已配置HADOOP_CONF_DIR指向YARN配置目录

核心配置文件conf/yarn-conf.yaml：

# YARN专属配置 yarn.application.name: flink-session-prod yarn.queue: default yarn.provided.lib.dirs: hdfs://namenode:8020/apps/flink/ # Flink资源配置（注意：这里配置的是Session申请的资源，不是单个TM） jobmanager.memory.process.size: 4g taskmanager.memory.process.size: 8g taskmanager.numberOfTaskSlots: 8 # 关键！必须指定YARN容器的JVM参数，否则容易OOM env.java.opts.jobmanager: "-XX:+UseG1GC -XX:MaxGCPauseMillis=50" env.java.opts.taskmanager: "-XX:+UseG1GC -XX:MaxGCPauseMillis=50" # 网络优化：YARN容器默认关闭hugepages，必须显式禁用 taskmanager.network.memory.off-heap: false

启动Session集群（生产环境必须用-nm指定NodeManager内存）：

# 申请一个带2个TaskManager的Session（每个TM 8G内存，共16G） ./bin/yarn-session.sh \ -jm 4096 \ -tm 8192 \ -s 8 \ -d \ # 后台运行 -nm flink-session-prod-20231001 \ -qu default \ -D yarn.application.name="Flink Prod Session" # 查看YARN上Application ID（格式类似application_1696123456789_0001） yarn application -list | grep flink

验证与调试技巧：

• 如果YARN Application状态一直是ACCEPTED，检查yarn.resourcemanager.scheduler.capacity.root.default.maximum-capacity是否设为100（默认是10，意味着最多用10%集群资源）；
• TaskManager启动失败常见原因是java.io.IOException: Failed to create directory: /tmp/flink-web-xxxx，需在yarn-site.xml里加yarn.nodemanager.env-whitelist=HADOOP_CONF_DIR,HADOOP_COMMON_HOME,...,FLINK_HOME；
• Web UI地址不是固定端口，而是YARN分配的随机端口，通过yarn application -status <appid>查看Tracking URL字段。

实操心得：

• 绝对不要在YARN Session里提交多个长期运行的作业！Session是共享资源池，一个作业的反压会拖垮所有作业。生产环境必须用Per-Job模式（每个作业独占一个YARN Application）；
• yarn-session.sh启动的Session默认不启用HA，必须加-D high-availability=zookeeper参数；
• YARN的日志聚合功能（yarn.log-aggregation-enable=true）必须开启，否则TaskManager崩溃后日志会随Container消失，只剩JobManager日志，根本没法查。

3.3 方案三：Kubernetes Native集群（云原生时代终极形态）

Kubernetes Native模式代表Flink集群的未来——JobManager和TaskManager作为原生K8s Workload运行，由K8s Controller统一管理生命周期。它彻底摆脱了YARN的抽象层，能充分利用K8s的Service Mesh、NetworkPolicy、HPA等能力。

核心K8s资源清单（flink-native-session.yaml）：

# JobManager StatefulSet（必须用StatefulSet保证稳定DNS） apiVersion: apps/v1 kind: StatefulSet metadata: name: flink-jobmanager spec: serviceName: flink-jobmanager replicas: 1 template: spec: containers: - name: jobmanager image: flink:1.17.1 env: - name: JOB_MANAGER_RPC_ADDRESS value: flink-jobmanager - name: FLINK_PROPERTIES value: | jobmanager.rpc.address: flink-jobmanager taskmanager.numberOfTaskSlots: 4 state.backend: rocksdb state.checkpoints.dir: s3://my-bucket/flink/checkpoints state.savepoints.dir: s3://my-bucket/flink/savepoints high-availability: kubernetes high-availability.storageDir: s3://my-bucket/flink/ha/ kubernetes.cluster-id: flink-native-prod ports: - containerPort: 6123 # RPC - containerPort: 6124 # BlobServer - containerPort: 8081 # WebUI resources: limits: memory: "4Gi" cpu: "2" requests: memory: "4Gi" cpu: "2" --- # TaskManager Deployment（用Deployment实现弹性扩缩） apiVersion: apps/v1 kind: Deployment metadata: name: flink-taskmanager spec: replicas: 3 selector: matchLabels: app: flink-taskmanager template: spec: containers: - name: taskmanager image: flink:1.17.1 env: - name: JOB_MANAGER_RPC_ADDRESS value: flink-jobmanager - name: FLINK_PROPERTIES value: | jobmanager.rpc.address: flink-jobmanager taskmanager.numberOfTaskSlots: 4 taskmanager.memory.process.size: 8g taskmanager.network.memory.fraction: 0.25 state.backend: rocksdb ports: - containerPort: 6122 # DataPort resources: limits: memory: "8Gi" cpu: "4" requests: memory: "8Gi" cpu: "4"

关键部署命令：

# 创建Namespace和ServiceAccount（必须） kubectl create namespace flink-prod kubectl create serviceaccount flink-service-account -n flink-prod kubectl create clusterrolebinding flink-rbac --clusterrole=edit --serviceaccount=flink-prod:flink-service-account # 应用资源清单 kubectl apply -f flink-native-session.yaml -n flink-prod # 检查Pod状态（JobManager必须Running后，TaskManager才能注册） kubectl get pods -n flink-prod # 获取WebUI地址（通过NodePort或Ingress暴露） kubectl port-forward svc/flink-jobmanager-rest 8081:8081 -n flink-prod

避坑指南：

• Kubernetes Native模式要求Flink版本≥1.15，且必须用官方Docker镜像（flink:1.17.1），自定义镜像需确保/opt/flink/conf/flink-conf.yaml被正确挂载；
• high-availability: kubernetes必须配合kubernetes.cluster-id，否则多个Flink集群会互相干扰；
• S3作为状态后端时，必须在flink-conf.yaml里配置AWS凭证（s3.access-key/s3.secret-key），或用IRSA（IAM Roles for Service Accounts）方式注入；
• TaskManager的replicas不能设为0，否则JobManager会一直等待Slot注册，状态卡在INITIALIZING。

实操心得：

• 生产环境务必给JobManager加livenessProbe和readinessProbe，例如：

  livenessProbe: httpGet: path: /jobmanager/config port: 8081 initialDelaySeconds: 60 periodSeconds: 30

• 使用kubectl scale deployment flink-taskmanager --replicas=5 -n flink-prod可动态扩缩TaskManager，Flink会自动感知新Slot；
• K8s的Eviction机制可能导致TaskManager被驱逐，必须在PodSpec里加priorityClassName: flink-high-priority并配置PriorityClass。

4. 集群健康度黄金指标监控与故障速查表

4.1 必须盯死的5个核心指标（附Prometheus查询语句）

Flink集群没有“健康”或“不健康”的二元状态，只有指标漂移程度。以下5个指标一旦越界，90%的线上问题都能提前预警：

实操技巧：把这5个查询保存为Grafana Dashboard的Alert Panel，设置for: 5m，一旦触发立即通知。我习惯把告警消息模板写成：“【Flink告警】${job} ${instance} ${metric} ${value}，请检查${runbook_url}”。Runbook URL指向内部Wiki的标准化排查手册，里面包含每种告警对应的kubectl logs、jstack、jmap命令集合。

4.2 故障速查表：从现象到根因的10分钟定位法

当集群出问题时，别急着重启，按这个流程走，10分钟内大概率定位：

独家经验：我在所有Flink集群的conf/log4j-console.properties里加了一行：

logger.checkpoint.name = org.apache.flink.runtime.checkpoint logger.checkpoint.level = INFO

这样Checkpoint关键事件（start/complete/abort）会单独打到console日志，kubectl logs -f jm-pod | grep checkpoint就能实时盯住，比翻完整日志快10倍。

4.3 日常巡检Checklist（运维同学必存）

每天早上花5分钟执行，能避免80%的半夜告警：

1. 资源水位检查
   kubectl top nodes确认CPU/Memory使用率 < 70%
   kubectl top pods -n flink-prod确认JM/TM内存使用 < 85%

2. Checkpoint健康检查
   curl http://jm:8081/jobs/overview | jq '.jobs[] | select(.status=="RUNNING") | .id'获取所有运行中jobId
   对每个jobId执行：curl "http://jm:8081/jobs/$id/checkpoints" | jq '.latestCheckpoint.status'，确保是COMPLETED

3. 网络连通性检查
   kubectl exec -it jm-pod -- ping -c 3 tm-pod-0
kubectl exec -it tm-pod-0 -- telnet jm-pod 6123（验证RPC端口）

4. 状态后端可用性检查
   kubectl exec -it jm-pod -- hdfs dfs -ls /flink/checkpoints/（HDFS）
   kubectl exec -it jm-pod -- aws s3 ls s3://bucket/flink/checkpoints/（S3）

5. 日志轮转检查
   kubectl exec -it jm-pod -- ls -lh /opt/flink/log/ | tail -5，确认最大日志文件 < 200MB，避免填满PV

注意：所有检查命令必须封装成Shell脚本，加入Cron定时任务。我用的脚本名叫flink-health-check.sh，每次执行后自动发企业微信消息，内容包含“✅ 全部通过”或“❌ 发现X个异常，请立即处理”。

5. 从“能跑”到“跑稳”的5个关键跃迁点

5.1 配置管理：告别vim conf/flink-conf.yaml的手工时代

手工改配置是集群不稳定的头号杀手。我见过最离谱的案例：某团队3个环境（dev/test/prod）共用同一份flink-conf.yaml，靠注释开关不同配置，结果一次上线把prod的state.backend从rocksdb改成了filesystem，导致所有Checkpoint写到本地磁盘，第二天集群重启后全量重放。

正确做法是引入配置中心：

• 中小团队：用GitOps + Kustomize。把base/目录放通用配置，overlays/prod/目录放生产专用patch，CI/CD流水线自动kubectl apply -k overlays/prod/；
• 大型企业：接入Apollo或Nacos。Flink 1.15+支持flink-conf.yaml里写${apollo.namespace.key}，启动时自动拉取；
• 云厂商客户：直接用阿里云ACM或腾讯云SSM，配置变更自动触发Flink Operator滚动更新。

关键原则：所有环境变量（如state.checkpoints.dir）必须从外部注入，flink-conf.yaml里只保留绝对不变的参数（如parallelism.default: 1）。

5.2 版本治理：为什么Flink 1.15到1.17的升级要拆成3步？

Flink版本升级不是wget换包那么简单。1.15→1.17跨越了3个大版本，API、序列化、状态后端都有破坏性变更：

• Step 1：先升到1.15.4（LTS）
  目标：验证Checkpoint兼容性。1.15开始RocksDB状态后端默认启用predefined-options: OPTIMIZED_FOR_LARGE_DATA，老版本Checkpoint可能加载失败。必须用flink savepoint upgrade命令迁移。

• Step 2：再升到1.16.3
  目标：适配新的Watermark机制。1.16废弃了AssignerWithPunctuatedWatermarks，所有自定义Watermark Assigner必须重构为WatermarkStrategy接口。

• Step 3：最后升到1.17.1
  目标：启用Native Kubernetes。1.17正式GA了kubernetes.operator，可以替换掉社区版Flink Operator。

血泪教训：升级前必须做三件事：

1. 用flink list -a导出所有运行中作业的JAR SHA256，备份到安全位置；
2. 在测试集群用--fromSavepoint参数从prod最新Savepoint启动，验证100%功能；
3. 准备回滚Plan：如果升级失败，用flink run -s <savepoint-path> -d命令降级启动。

5.3 安全加固：生产环境必须关闭的3个默认开关

Flink默认配置为了开发便利，牺牲了安全性。生产环境上线前，这3个开关必须关：

1. 关闭REST API未授权访问
   默认rest.address: 0.0.0.0且无认证。必须加：

   rest.authentication.enabled: true rest.authentication.signing-key: "your-secret-key-here"

2. 禁用BlobServer文件上传
   blob.server.port默认开放，攻击者可上传