且解析正确,可以ping通

发布时间:2025-06-24 17:53:04  作者:北方职教升学中心  阅读量:331


  • 解决方案

2种:

  1. 使用支持"UseG1GC"参数的jdk镜像。

    3.2 Service

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

    3.4 jobmanager

    jobmanager-session-deployment.yaml

    这里我选择使用session模式。还是优先使用配置文件,我记不清了...

    但是,当你启用了配置文件可写、

  2. 这里引入了1个初始化容器(initContainers),用来创建目录、

    为保证集群能够运行起来,对文件进行了调整。

    4.1.1 配置文件不可写

    • 原因

    taskmanager采用官方的配置会出现这个问题。且解析正确,可以ping通。

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

    但会影响rpc通信(后面会有解析)。

    关键配置:

    引入了初始化容器,将ConfigMap中的配置文件拷贝到emptyDir中的/opt/flink/conf/路径下。

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

    apiVersion: v1kind: ConfigMapmetadata:  name: flink-config  namespace: "flink-ns"  labels:    app: flinkdata:  flink-conf.yaml: |+    jobmanager.rpc.address: flink-jobmanager    taskmanager.numberOfTaskSlots: 2    blob.server.port: 6124    jobmanager.rpc.port: 6123    taskmanager.rpc.port: 6122    jobmanager.memory.process.size: 1600m    taskmanager.memory.process.size: 1728m    parallelism.default: 2  log4j-console.properties: |+    rootLogger.level = INFO    rootLogger.appenderRef.console.ref = ConsoleAppender    rootLogger.appenderRef.rolling.ref = RollingFileAppender    logger.flink.name = org.apache.flink    logger.flink.level = INFO    logger.pekko.name = org.apache.pekko    logger.pekko.level = INFO    logger.kafka.name= org.apache.kafka    logger.kafka.level = INFO    logger.hadoop.name = org.apache.hadoop    logger.hadoop.level = INFO    logger.zookeeper.name = org.apache.zookeeper    logger.zookeeper.level = INFO    # 将所有 info 级别的日志输出到 console    appender.console.name = ConsoleAppender    appender.console.type = CONSOLE    appender.console.layout.type = PatternLayout    appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n    # 将所有 info 级别的日志输出到指定的 rolling file    appender.rolling.name = RollingFileAppender    appender.rolling.type = RollingFile    appender.rolling.append = false    appender.rolling.fileName = ${sys:log.file}    appender.rolling.filePattern = ${sys:log.file}.%i    appender.rolling.layout.type = PatternLayout    appender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n    appender.rolling.policies.type = Policies    appender.rolling.policies.size.type = SizeBasedTriggeringPolicy    appender.rolling.policies.size.size=100MB    appender.rolling.strategy.type = DefaultRolloverStrategy    appender.rolling.strategy.max = 10    # 关闭 Netty channel handler 中不相关的(错误)警告    logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline    logger.netty.level = OFF

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

    jobmanager.rpc.address: flink-jobmanager

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

    flink-jobmanager,该服务需要另行定义。且没有设置环境变量时,此种情况必然会出现。在此基础上修改yaml文件,设置环境变量 JOB_MANAGER_RPC_ADDRESS ,值设置为rpc的service名称:

    JOB_MANAGER_RPC_ADDRESS 设置

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

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

    4.1.2 VM启动失败

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

    基本不用动,可以直接拷贝过来使用(在生产环境中需要根据实际情况修改配置参数)。编辑、

    5 复盘

    历尽艰辛算是部署成功了。

    • 解决方案

    将ConfigMap中的文件copy出来,挂载到本地目录上;主容器在启动时挂载本地目录即可。这里复盘一下。当使用官方的 taskmanager-session-deployment.yaml 配置时,可能会出现无法连接到jobmanager的情况。测试环境上够用,但上生产还是有段路要走的。

    3.2.2 rest

    jobmanager-rest-service.yaml

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

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

    k8s在启动taskmanager容器启动时,会将ConfigMap中的配置信息拷贝到容器中,而这些信息是只读的。当然也可以挂到pv上,但配置都在ConfigMap中维护,没必要外接pv。

    apiVersion: v1kind: Servicemetadata:  name: flink-jobmanager-rest  namespace: "flink-ns"spec:  type: NodePort  ports:    - name: rest      port: 8081      targetPort: 8081      nodePort: 30081  selector:    app: flink    component: jobmanager

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

    在启动taskmanager时,会修改/opt/flink/conf/flink-config.yaml文件(见docker-entrypoint.sh文件),使用环境变量来更新配置。

    • 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、里面的干扰项如图:

    官方 jobmanager-session-deployment-ha.yaml

    ha模式在启动时,使用POD_IP这个环境变量指定了 jobmanager.rpc.address 参数为当前Pod的IP地址,而没有使用配置文件中的配置。

    3.1 ConfigMap

    flink-configuration-configmap.yaml

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

    3.5 taskmanager

    taskmanager-session-deployment.yaml

    该文件的内容如下:

    apiVersion: apps/v1kind: Deploymentmetadata:  name: flink-taskmanager  namespace: "flink-ns"spec:  replicas: 3  selector:    matchLabels:      app: flink      component: taskmanager  template:    metadata:      labels:        app: flink        component: taskmanager    spec:      initContainers:        - name: init-flink-conf          image: flink-arm:latest          command: [ "sh", "-c", "mkdir -p /opt/flink/conf /config ; cp /config/* /opt/flink/conf ; chmod 777 /opt/flink/conf/*" ]          volumeMounts:            - name: flink-config-volume              mountPath: /config            - name: local-storage              mountPath: /opt/flink/conf      containers:        - name: taskmanager          image: flink-arm:latest          args: [ "taskmanager" ]          ports:            - containerPort: 6122              name: rpc          livenessProbe:            tcpSocket:              port: 6122            initialDelaySeconds: 30            periodSeconds: 60          volumeMounts:            - name: local-storage              mountPath: /opt/flink/conf          env:            - name: JOB_MANAGER_RPC_ADDRESS              value: flink-jobmanager          securityContext:            runAsUser: 9999  # 参考官方 flink 镜像中的 _flink_ 用户,如有必要可以修改      volumes:        - name: flink-config-volume          configMap:            name: flink-config            items:              - key: flink-conf.yaml                path: flink-conf.yaml              - key: log4j-console.properties                path: log4j-console.properties        - name: local-storage          emptyDir: { }

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

    1. 引入了emptyDir,用于存放配置文件。我遇到了2种。删除 ConfigMap 权限的 Service 账号 volumes: - name: flink-config-volume configMap: name: flink-config items: - key: flink-conf.yaml path: flink-conf.yaml - key: log4j-console.properties path: log4j-console.properties

      说明:

      原本选择了ha模式的yaml文件,但部署过程中遇到了一些问题。这导致了 jobmanager 和 tarskmanager 两边的 jobmanager.rpc.address 配置不同,jobmanager 就将 tarskmanager 消息丢弃。

      说明:

      taskmanager启动时是优先使用环境变量来配置、

      2 准备工作

      flink的arm镜像

      flink的arm镜像:

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

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

      查看端口状态,6123是连接状态。

      apiVersion: v1kind: Servicemetadata:  name: flink-jobmanager  labels:    app: flink    component: jobmanagerspec:  type: ClusterIP  clusterIP: None  selector:    app: flink    component: jobmanager  ports:  - name: rpc    port: 6123

      这里:

      metadata.name 需要与 ConfigMap 中的 jobmanager.rpc.address 参数一致。结果因为这个浪费了大半天时间。

      注意:

      这个问题正常情况下不会影响taskmanager的启动,只是配置无法更新。说明连接已建立,也进行了rpc通信,只不过jobmanager不承认tarskmanager!

      • 原因

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

      在我的案例中,我一开始想部署session-ha模式(高估了自己),jobmanager 使用了错误的yaml。但把消息丢弃了。

      5.1 ConfigMap是只读的

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

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

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

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

    4 问题及解决方案

    起初我按照官方的配置来部署,出现了很多问题。# args: [ "jobmanager", "$(POD_IP)" ] args: [ "jobmanager"] ports: - containerPort: 6123 name: rpc - containerPort: 6124 name: blob-server - containerPort: 8081 name: webui livenessProbe: tcpSocket: port: 6123 initialDelaySeconds: 30 periodSeconds: 60 volumeMounts: - name: flink-config-volume mountPath: /opt/flink/conf securityContext: runAsUser: 9999 # 参考官方 flink 镜像中的 _flink_ 用户,如有必要可以修改 serviceAccountName: flink-service-account # 拥有创建、

    1 背景

    目标:在arm架构的K8S上部署一套flink集群。并将配置文件复制到emptyDir中的/opt/flink/conf路径下。本文记录在此过程中遇到的问题,以及问题原因和解决方案。无法连接 flink-jobmanager;而 jobmanager 收到了来自 tarskmanager 的消息(172.224.135.172 为172.224.135.172实例的地址)、

    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 这个配置。官方的配置文件经过调教后如下:

    apiVersion: apps/v1kind: Deploymentmetadata:  name: flink-jobmanager  namespace: "spec:  replicas: 1 # 通过设置大于 1 的整型值来开启 Standby JobManager  selector:    matchLabels:      app: flink      component: jobmanager  template:    metadata:      labels:        app: flink        component: jobmanager    spec:      containers:        - name: jobmanager          image: flink-arm:lastest#          env:#            - name: POD_IP#              valueFrom:#                fieldRef:#                  apiVersion: v1#                  fieldPath: status.podIP#          # 下面的 args 参数会使用 POD_IP 对应的值覆盖 config map 中 jobmanager.rpc.address 的属性值。

    • 解决

    完成4.1.1,使配置文件可写。删除 ConfigMap 权限的 Service 账号。

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

    • 解决

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

    3.3 ServiceAccount

    flink-service-account.yaml

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

    apiVersion: v1kind: ServiceAccountmetadata:  name: flink-service-account  namespace: "flink-ns"---apiVersion: rbac.authorization.k8s.io/v1kind: Rolemetadata:  name: configmap-managerrules:  - apiGroups: [ "" ]    resources: [ "configmaps" ]    verbs: [ "create", "delete", "get", "list", "patch", "update", "watch" ]---apiVersion: rbac.authorization.k8s.io/v1kind: RoleBindingmetadata:  name: configmap-manager-binding  namespace: "flink-ns"subjects:  - kind: ServiceAccount    name: flink-service-accountroleRef:  kind: Role  name: configmap-manager  apiGroup: rbac.authorization.k8s.io

    文件定义了拥有创建、编辑、但这可能破坏了ha特性,有待验证。这里即便你在容器中使用root启动,还是没有写的权限。

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

    3 部署

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

    我在 tarskmanager的 pod中,执行 ping flink-jobmanager ,可以正常解析 flink-jobmanager 的ip地址、

    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 systemsed: 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 systemStarting Task ManagerStarting 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启动失败

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

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

    也许你也会踩一遍我经历过的坑。修改权限、

    5.3 flink-on-k8s ha模式

    一开始如果乖乖的使用非ha模式,应该不会有4.2.2的问题。

    我对k8s还算了解,但在此之前没接触过flink,部署起来确实有点困难。

    3.2.1 rpc

    flink-rpc-service.yaml

    官方未提供该文件,这里定义了rpc通信的端口,taskmanager可以在集群中通过服务名flink-jobmanager找到jobmanager。docker-entrypoint.sh 关键截图:

    JOB_MANAGER_RPC_ADDRESS定义

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

    更新配置文件flink-config.yaml

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

    日志中提到:

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

    这是taskmanager.sh中报的错误:

    taskmanager.sh截图

    • 原因

    这是因为我用的基础镜像jdk不支持UseG1GC,需要显示启用。taskmanager启动时仍会读取默认的配置(即ConfigMap)。

  3. 在镜像构建时修改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通信。

  • 设置了环境变量JOB_MANAGER_RPC_ADDRESS,指定了rpc的服务地址是flink-jobmanager。

    根据现象来看,环境变量的优先级是高于配置文件的,前提是你的配置文件要可写。容器挂载的目录不再是ConfigMap,而是emptyDir。

    k8s环境

    已有k8s集群环境。但2个服务的日志中有报错。

    4.1 taskmanager启动失败

    现象:jobmanager正常启动,但taskmanager启动失败。那么问题来了:

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

    有待验证...