Flink(arm) on K8S 部署时的那些坑

1 背景

目标：在arm架构的K8S上部署一套flink集群。

我对k8s还算了解，但在此之前没接触过flink，部署起来确实有点困难。本文记录在此过程中遇到的问题，以及问题原因和解决方案。

2 准备工作

flink的arm镜像

flink的arm镜像：

1. 需包含arm版jdk1.8
2. flink版本为1.17.2

该镜像的构建方法可参考我的文章：arm环境下构建Flink的Docker镜像。

k8s环境

已有k8s集群环境。

我所在的环境是1.18，部署在arm架构的服务器上。

3 部署

我参考了官方的经验（见 https://nightlies.apache.org/flink/flink-docs-master/zh/docs/deployment/resource-providers/standalone/kubernetes/ ），建议大家酌情参考。

也许你也会踩一遍我经历过的坑。

3.1 ConfigMap

flink-configuration-configmap.yaml

这个是flink的配置文件，对应着安装包中 ${FLINK_HOME}/conf/flink-conf.yaml。

基本不用动，可以直接拷贝过来使用（在生产环境中需要根据实际情况修改配置参数）。

apiVersion: v1
kind: ConfigMap
metadata:name: flink-confignamespace: "flink-ns"labels:app: flink
data:flink-conf.yaml: |+jobmanager.rpc.address: flink-jobmanagertaskmanager.numberOfTaskSlots: 2blob.server.port: 6124jobmanager.rpc.port: 6123taskmanager.rpc.port: 6122jobmanager.memory.process.size: 1600mtaskmanager.memory.process.size: 1728mparallelism.default: 2log4j-console.properties: |+rootLogger.level = INFOrootLogger.appenderRef.console.ref = ConsoleAppenderrootLogger.appenderRef.rolling.ref = RollingFileAppenderlogger.flink.name = org.apache.flinklogger.flink.level = INFOlogger.pekko.name = org.apache.pekkologger.pekko.level = INFOlogger.kafka.name= org.apache.kafkalogger.kafka.level = INFOlogger.hadoop.name = org.apache.hadooplogger.hadoop.level = INFOlogger.zookeeper.name = org.apache.zookeeperlogger.zookeeper.level = INFO# 将所有 info 级别的日志输出到 consoleappender.console.name = ConsoleAppenderappender.console.type = CONSOLEappender.console.layout.type = PatternLayoutappender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n# 将所有 info 级别的日志输出到指定的 rolling fileappender.rolling.name = RollingFileAppenderappender.rolling.type = RollingFileappender.rolling.append = falseappender.rolling.fileName = ${sys:log.file}appender.rolling.filePattern = ${sys:log.file}.%iappender.rolling.layout.type = PatternLayoutappender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%nappender.rolling.policies.type = Policiesappender.rolling.policies.size.type = SizeBasedTriggeringPolicyappender.rolling.policies.size.size=100MBappender.rolling.strategy.type = DefaultRolloverStrategyappender.rolling.strategy.max = 10# 关闭 Netty channel handler 中不相关的（错误）警告logger.netty.name = org.jboss.netty.channel.DefaultChannelPipelinelogger.netty.level = OFF

配置文件中，部署集群时需要重点关注的参数：

jobmanager.rpc.address: flink-jobmanager

该参数定义了 taskmanager 与 jobmanager 通信的rpc地址，这里使用了服务名称

flink-jobmanager，该服务需要另行定义。

3.2 Service

为清晰管理服务，我将rpc服务和rest服务拆开了。

3.2.1 rpc

flink-rpc-service.yaml

官方未提供该文件，这里定义了rpc通信的端口，taskmanager可以在集群中通过服务名flink-jobmanager找到jobmanager。

apiVersion: v1
kind: Service
metadata:name: flink-jobmanagerlabels:app: flinkcomponent: jobmanager
spec:type: ClusterIPclusterIP: Noneselector:app: flinkcomponent: jobmanagerports:- name: rpcport: 6123

这里：

metadata.name 需要与 ConfigMap 中的 jobmanager.rpc.address 参数一致。

spec.ports.port 需要与 ConfigMap 中的 jobmanager.rpc.port 参数一致。

3.2.2 rest

jobmanager-rest-service.yaml

该文件定义了jobmanager控制台暴露的web端口，在官方文档中有提供，直接拷贝过来即可。

apiVersion: v1
kind: Service
metadata:name: flink-jobmanager-restnamespace: "flink-ns"
spec:type: NodePortports:- name: restport: 8081targetPort: 8081nodePort: 30081selector:app: flinkcomponent: jobmanager

部署完成后，可以通过 8081 端口访问控制台。

3.3 ServiceAccount

flink-service-account.yaml

该文件官方未提供，但在jobmanager中有引用，需要自行定义：

apiVersion: v1
kind: ServiceAccount
metadata:name: flink-service-accountnamespace: "flink-ns"
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:name: configmap-manager
rules:- apiGroups: [ "" ]resources: [ "configmaps" ]verbs: [ "create", "delete", "get", "list", "patch", "update", "watch" ]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:name: configmap-manager-bindingnamespace: "flink-ns"
subjects:- kind: ServiceAccountname: flink-service-account
roleRef:kind: Rolename: configmap-managerapiGroup: rbac.authorization.k8s.io

文件定义了拥有创建、编辑、删除 ConfigMap 权限的 Service 账号。

3.4 jobmanager

jobmanager-session-deployment.yaml

这里我选择使用session模式。官方的配置文件经过调教后如下：

apiVersion: apps/v1
kind: Deployment
metadata:name: flink-jobmanagernamespace: "
spec:replicas: 1 # 通过设置大于 1 的整型值来开启 Standby JobManagerselector:matchLabels:app: flinkcomponent: jobmanagertemplate:metadata:labels:app: flinkcomponent: jobmanagerspec:containers:- name: jobmanagerimage: flink-arm:lastest
#          env:
#            - name: POD_IP
#              valueFrom:
#                fieldRef:
#                  apiVersion: v1
#                  fieldPath: status.podIP
#          # 下面的 args 参数会使用 POD_IP 对应的值覆盖 config map 中 jobmanager.rpc.address 的属性值。
#          args: [ "jobmanager", "$(POD_IP)" ]args: [ "jobmanager"]ports:- containerPort: 6123name: rpc- containerPort: 6124name: blob-server- containerPort: 8081name: webuilivenessProbe:tcpSocket:port: 6123initialDelaySeconds: 30periodSeconds: 60volumeMounts:- name: flink-config-volumemountPath: /opt/flink/confsecurityContext:runAsUser: 9999  # 参考官方 flink 镜像中的 _flink_ 用户，如有必要可以修改serviceAccountName: flink-service-account # 拥有创建、编辑、删除 ConfigMap 权限的 Service 账号volumes:- name: flink-config-volumeconfigMap:name: flink-configitems:- key: flink-conf.yamlpath: flink-conf.yaml- key: log4j-console.propertiespath: log4j-console.properties

说明：

原本选择了ha模式的yaml文件，但部署过程中遇到了一些问题。

为保证集群能够运行起来，对文件进行了调整。但这可能破坏了ha特性，有待验证。

3.5 taskmanager

taskmanager-session-deployment.yaml

该文件的内容如下：

apiVersion: apps/v1
kind: Deployment
metadata:name: flink-taskmanagernamespace: "flink-ns"
spec:replicas: 3selector:matchLabels:app: flinkcomponent: taskmanagertemplate:metadata:labels:app: flinkcomponent: taskmanagerspec:initContainers:- name: init-flink-confimage: flink-arm:latestcommand: [ "sh", "-c", "mkdir -p /opt/flink/conf /config ; cp /config/* /opt/flink/conf ; chmod 777 /opt/flink/conf/*" ]volumeMounts:- name: flink-config-volumemountPath: /config- name: local-storagemountPath: /opt/flink/confcontainers:- name: taskmanagerimage: flink-arm:latestargs: [ "taskmanager" ]ports:- containerPort: 6122name: rpclivenessProbe:tcpSocket:port: 6122initialDelaySeconds: 30periodSeconds: 60volumeMounts:- name: local-storagemountPath: /opt/flink/confenv:- name: JOB_MANAGER_RPC_ADDRESSvalue: flink-jobmanagersecurityContext:runAsUser: 9999  # 参考官方 flink 镜像中的 _flink_ 用户，如有必要可以修改volumes:- name: flink-config-volumeconfigMap:name: flink-configitems:- key: flink-conf.yamlpath: flink-conf.yaml- key: log4j-console.propertiespath: log4j-console.properties- name: local-storageemptyDir: { }

该文件与官方提供的有些许不同：

1. 引入了emptyDir，用于存放配置文件。
2. 这里引入了1个初始化容器（initContainers），用来创建目录、修改权限、并将配置文件复制到emptyDir中的/opt/flink/conf路径下。
3. 设置了环境变量JOB_MANAGER_RPC_ADDRESS，指定了rpc的服务地址是flink-jobmanager。

4 问题及解决方案

起初我按照官方的配置来部署，出现了很多问题。

下面是遇到的典型问题以及解决方案。

4.1 taskmanager启动失败

现象：jobmanager正常启动，但taskmanager启动失败。

taskmanager日志中报错：

WARNING: attempted to load jemalloc from /usr/lib/aarch64-linux-gnu/libjemalloc.so and /usr/lib/x86_64-linux-gnu/libjemalloc.so but the library couldn't be found. glibc will be used instead.
sed: couldn't open temporary file /opt/flink/conf/sedGGEvws: Read-only file system
sed: couldn't open temporary file /opt/flink/conf/sedL9ELTs: Read-only file system
/docker-entrypoint.sh: line 73: /opt/flink/conf/flink-conf.yaml: Permission denied
/docker-entrypoint.sh: line 89: /opt/flink/conf/flink-conf.yaml.tmp: Read-only file system
Starting Task Manager
Starting taskexecutor as a console application on host flink-taskmanager-b6c9fbb8c-hxc4h.
Error: VM option 'UseG1GC' is experimental and must be enabled via -XX:+UnlockExperimentalVMOptions.
Error: Could not create the Java Virtual Machine.
Error: A fatal exception has occurred. Program will exit.

从日志中可以发现2个问题：

1. 配置文件不可写
2. VM启动失败

下面分别说一下解决方案。

4.1.1 配置文件不可写

• 原因

taskmanager采用官方的配置会出现这个问题。官方配置如下图：

taskmanager-session-deployment.yaml中的配置文件挂载

k8s在启动taskmanager容器启动时，会将ConfigMap中的配置信息拷贝到容器中，而这些信息是只读的。

在启动taskmanager时，会修改/opt/flink/conf/flink-config.yaml文件（见docker-entrypoint.sh文件），使用环境变量来更新配置。这里即便你在容器中使用root启动，还是没有写的权限。

• 解决方案

将ConfigMap中的文件copy出来，挂载到本地目录上；主容器在启动时挂载本地目录即可。当然也可以挂到pv上，但配置都在ConfigMap中维护，没必要外接pv。

我的案例中，将配置文件copy到了emptyDir中，这时taskmanager就可以写配置文件了。

关键配置：

引入了初始化容器，将ConfigMap中的配置文件拷贝到emptyDir中的/opt/flink/conf/路径下。容器挂载的目录不再是ConfigMap，而是emptyDir。

注意：

这个问题正常情况下不会影响taskmanager的启动，只是配置无法更新。taskmanager启动时仍会读取默认的配置（即ConfigMap）。

但会影响rpc通信（后面会有解析）。

4.1.2 VM启动失败

这个问题是导致taskmanager启动失败的真正原因。

日志中提到：

Error: VM option 'UseG1GC' is experimental and must be enabled via -XX:+UnlockExperimentalVMOptions.

这是taskmanager.sh中报的错误：

taskmanager.sh截图

• 原因

这是因为我用的基础镜像jdk不支持UseG1GC，需要显示启用。

• 解决方案

2种：

1. 使用支持"UseG1GC"参数的jdk镜像。
2. 在镜像构建时修改taskmanager.sh文件：在Dockerfile中加入下面一行：

# 修改 taskmanager 启动参数
RUN sed -i 's/-XX:+UseG1GC/-XX:+UnlockExperimentalVMOptions -XX:+UseG1GC/g' bin/taskmanager.sh

4.2 taskmanager无法注册到jobmanager

taskmanager需通过flink-jobmanager:6123访问jobmanager，实现rpc通信。当使用官方的 taskmanager-session-deployment.yaml 配置时，可能会出现无法连接到jobmanager的情况。我遇到了2种。

4.2.1 rpc时连接到了本地容器

报错可能类似下方：

Could not connect to rpc endpoint under address akka.tcp://flink@flink-taskmanager-b6c9fbb8c-hxc4h:6123/user/rpc/resourcemanager_*.

flink-taskmanager-b6c9fbb8c-hxc4h 是我本地容器的实例名，怎么连接到了本地？

• 原因

这是因为在1.17.2版本的docker-entrypoint.sh中，优先使用环境变量 JOB_MANAGER_RPC_ADDRESS 来覆盖 jobmanager.rpc.address 这个配置。docker-entrypoint.sh 关键截图：

JOB_MANAGER_RPC_ADDRESS定义

如图，JOB_MANAGER_RPC_ADDRESS如果没有设置，会默认取本机主机名！

更新配置文件flink-config.yaml

在prepare_configuration() 中用环境变量更新配置文件中的 jobmanager.rpc.address值。

说明：

taskmanager启动时是优先使用环境变量来配置、还是优先使用配置文件，我记不清了...

但是，当你启用了配置文件可写、且没有设置环境变量时，此种情况必然会出现。

• 解决

完成4.1.1，使配置文件可写。在此基础上修改yaml文件，设置环境变量 JOB_MANAGER_RPC_ADDRESS ，值设置为rpc的service名称：

JOB_MANAGER_RPC_ADDRESS 设置

4.2.2 连接到了jobmanager，但消息被丢弃

这个情况比较极端，大家可能没有碰到过。

我在 tarskmanager的 pod中，执行 ping flink-jobmanager ，可以正常解析 flink-jobmanager 的ip地址、且解析正确，可以ping通。但2个服务的日志中有报错。

• jobmanager 报错：

2024-07-29 11:22:33,487 ERROR akka.remote.EndpointWriter [] - dropping message [class akka.actor.ActorSelectionMessage] for non-local recipient [Actor[akka.tcp://flink@flink-jobmanager:6123/]] arriving at [akka.tcp://flink@flink-jobmanager:6123] inbound addresses are [akka.tcp://flink@172.224.135.172:6123]

• tarskmanager 日志：

2024-07-29 11:22:23,464 INFO org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Could not resolve ResourceManager address akka.tcp://flink@flink-jobmanager:6123/user/rpc/resourcemanager_*, retrying in 10000 ms: Could not connect to rpc endpoint under address akka.tcp://flink@flink-jobmanager:6123/user/rpc/resourcemanager_*.

从日志分析，tarskmanager 无法解析 ResourceManager、无法连接 flink-jobmanager；而 jobmanager 收到了来自 tarskmanager 的消息（172.224.135.172 为172.224.135.172实例的地址）、但把消息丢弃了。

查看端口状态，6123是连接状态。说明连接已建立，也进行了rpc通信，只不过jobmanager不承认tarskmanager！

• 原因

在启动时，jobmanager 和 tarskmanager 对于 jobmanager.rpc.address 的配置必须相同！

在我的案例中，我一开始想部署session-ha模式（高估了自己），jobmanager 使用了错误的yaml。里面的干扰项如图：

官方 jobmanager-session-deployment-ha.yaml

ha模式在启动时，使用POD_IP这个环境变量指定了 jobmanager.rpc.address 参数为当前Pod的IP地址，而没有使用配置文件中的配置。这导致了 jobmanager 和 tarskmanager 两边的 jobmanager.rpc.address 配置不同，jobmanager 就将 tarskmanager 消息丢弃。

• 解决

去掉PodIP这个启动参数即可（详见3.4章节配置）。

5 复盘

历尽艰辛算是部署成功了。测试环境上够用，但上生产还是有段路要走的。这里复盘一下。

5.1 ConfigMap是只读的

以前对这个印象并不是很清晰，经历了这次有了深刻体会。

也让我学到了：ConfigMap也可以作为配置模板，然后通过本文中提到的挂载方式，在程序启动时动态更新配置。

5.2 flink中的环境变量与配置文件

没看过flink的源码，不知道哪个优先级更高。

根据现象来看，环境变量的优先级是高于配置文件的，前提是你的配置文件要可写。

5.3 flink-on-k8s ha模式

一开始如果乖乖的使用非ha模式，应该不会有4.2.2的问题。结果因为这个浪费了大半天时间。

再回来看看ha模式：jobmanager在启动时指定了PodIP，这说明多个 jobmanager 同时存在时，只有1个 jobmanager 接收 taskmanager 的注册信息。那么问题来了：

1. 这时taskmanager如何设置rpc地址为这个Pod的地址？
2. 多个jobmanager是如何同步注册信息的？

有待验证...