当前位置： 首页 > news >正文

网站怎么做搜索引擎优化、,网站后台网址后缀,郑州网站开发网站开发,新手如何制作网站引言 在当今大数据与云计算蓬勃发展的时代#xff0c;容器编排与流处理技术成为企业数据处理架构的关键支柱。Kubernetes作为容器编排系统的行业标准#xff0c;能够高效自动化地部署、扩展和管理计算机应用程序#xff1b;Apache Flink则是流处理和批处理领域的佼佼者容器编排与流处理技术成为企业数据处理架构的关键支柱。Kubernetes作为容器编排系统的行业标准能够高效自动化地部署、扩展和管理计算机应用程序Apache Flink则是流处理和批处理领域的佼佼者以强大的实时处理能力和精准的状态管理著称。当Flink与Kubernetes实现深度集成二者优势互补为企业带来了更加灵活、高效、智能的数据处理解决方案。本文将基于详实的操作笔记深入解析Flink与Kubernetes集成的全流程以及不同部署模式的技术要点与实践细节。 一、集成环境搭建 1.1 环境要求 Flink与Kubernetes的集成对运行环境有着严格且明确的要求这些条件是确保集成顺利进行和集群稳定运行的基础 Kubernetes版本Kubernetes集群版本必须在1.9及以上高版本的Kubernetes不仅提供了更丰富的功能特性还能更好地与Flink进行兼容性适配保证Flink作业在集群上的稳定运行和高效调度。kubectl配置kubecconfig文件是连接本地客户端与Kubernetes集群的桥梁通常存储在~/.kube/config路径下 。通过执行kubectl auth can-i list|create|edit|delete pods命令可以验证当前用户是否具备对Pods和服务进行列出、创建、编辑、删除等操作的权限。若权限不足在后续部署Flink集群资源时将会遇到权限拒绝的错误导致部署失败。Kubernetes DNSKubernetes DNS服务的开启至关重要它承担着集群内服务发现的核心功能。在Flink与Kubernetes集成环境中各组件之间需要通过服务名称进行通信Kubernetes DNS能够将服务名称解析为对应的IP地址确保Flink JobManager、TaskManager等组件之间的网络通信畅通无阻。RBAC权限默认的default服务账号需要具备RBAC基于角色的访问控制中创建、删除Pods的权限。然而为了实现更精细化的资源管理和权限隔离建议专门为Flink创建独立的命名空间和服务账号。这样做不仅可以避免因权限混乱导致的部署失败风险还能显著降低后期的运维成本使得Flink集群的管理更加清晰、安全和高效。 1.2 创建专属资源 为了给Flink集群打造一个独立、安全且便于管理的运行环境需要按照以下步骤创建专属资源 创建命名空间使用命令kubectl create ns flink即可创建一个名为flink的命名空间。这个命名空间就像是一个独立的“数据处理小世界”后续所有与Flink相关的资源如Pods、Services、ConfigMaps等都将部署在这个空间内实现与其他应用资源的逻辑隔离。创建ServiceAccount执行kubectl create serviceaccount flink-service-account -n flink命令创建flink-service-account服务账号。该服务账号将作为Flink集群与Kubernetes集群进行交互的“身份凭证”用于验证操作权限和进行安全通信。用户授权通过kubectl create clusterrolebinding flink-role-binding-flink –clusterroleedit –serviceaccountflink:flink-service-account命令为flink-service-account服务账号赋予edit权限。这意味着该账号可以对集群内的资源进行创建、修改和删除等操作从而满足Flink集群部署和运行过程中对资源管理的需求 。 二、Flink Standalone Kubernetes部署模式 2.1 模式概述 Flink Standalone Kubernetes集成模式支持session和application两种部署模式而per-job模式目前仅在YARN环境中支持并且在Flink 1.15版本中已在YARN环境下被弃用具体可参考FLINK-26000相关内容 。本次重点研究非HA高可用部署模式虽然在实际生产环境中HA模式能够提供更高的可靠性和容错能力但非HA模式与HA模式在核心原理和大部分配置上是相通的HA模式的详细配置可参考官方文档进行深入学习。session和application模式的主要差异体现在JobManager和TaskManager服务的声明方式上不过两种模式也存在一些通用的集群资源配置。 2.2 通用集群资源配置 在Flink Standalone Kubernetes部署中以下几种资源是通用的它们为Flink集群的正常运行提供了基础配置和服务支持 flink-configuration-configmap.yaml该文件主要用于配置Flink的核心参数以及日志相关设置是Flink集群运行的重要配置文件。示例内容如下 apiVersion: v1 kind: ConfigMap metadata:name: flink-configlabels:app: flink data:flink-conf.yaml: |jobmanager.rpc.address: flink-jobmanagertaskmanager.numberOfTaskSlots: 2blob.server.port: 6124jobmanager.rpc.port: 6123taskmanager.rpc.port: 6122queryable-state.proxy.ports: 6125jobmanager.memory.process.size: 1600mtaskmanager.memory.process.size: 1728mparallelism.default: 2 log4j-console.properties: |# 如下配置会同时影响用户代码和 Flink 的日志行为rootLogger.level INFOrootLogger.appenderRef.console.ref ConsoleAppenderrootLogger.appenderRef.rolling.ref RollingFileAppender# 如果你只想改变 Flink 的日志行为则可以取消如下的注释部分#logger.flink.name org.apache.flink#logger.flink.level INFO# 下面几行将公共 libraries 或 connectors 的日志级别保持在 INFO 级别。# root logger 的配置不会覆盖此处配置。# 你必须手动修改这里的日志级别。logger.akka.name akkalogger.akka.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.size100MBappender.rolling.strategy.type DefaultRolloverStrategyappender.rolling.strategy.max 10# 关闭 Netty channel handler 中不相关的错误警告logger.netty.name org.jboss.netty.channel.DefaultChannelPipelinelogger.netty.level OFF flink-reactive-mode-configuration-configmap.yaml当需要启用Flink的响应式调度模式时需要配置此文件。除了包含与flink-configuration-configmap.yaml类似的核心参数配置外还会设置与响应式模式相关的特定参数如调度模式和检查点间隔等。示例如下 apiVersion: v1 kind: ConfigMap metadata:name: flink-configlabels: app: flink data:flink-conf.yaml: jobmanager.rpc.address: flink-jobmanagertaskmanager.numberOfTaskSlots: 2 blob.server.port: 6124 jobmanager.rpc.port: 6123 taskmanager.rpc.port: 6122 queryable-state.proxy.ports: 6125 jobmanager.memory.process.size: 1600m taskmanager.memory.process.size: 1728m parallelism.default: 2scheduler-mode: reactiveexecution.checkpointing.interval: 10s log4j-console.properties: |# 如下配置会同时影响用户代码和 Flink 的日志行为rootLogger.level INFOrootLogger.appenderRef.console.ref ConsoleAppender rootLogger.appenderRef.rolling.ref RollingFileAppender# 如果你只想改变 Flink 的日志行为则可以取消如下的注释部分#logger.flink.name org.apache.flink #logger.flink.level INFO # 下面几行将公共 libraries 或 connectors 的日志级别保持在 INFO 级别。# root logger 的配置不会覆盖此处配置。# 你必须手动修改这里的日志级别。logger.akka.name akkalogger.akka.level INFOlogger.kafka.name org.apache.kafka logger.kafka.level INFOlogger.hadoop.name org.apache.hadooplogger.hadoop.level INFO logger.zookeeper.name org.apache.zookeeper logger.zookeeper.level INFO # 将所有 info 级别的日志输出到 consoleappender.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 fileappender.rolling.name RollingFileAppenderappender.rolling.type RollingFileappender.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.size100MB appender.rolling.strategy.type DefaultRolloverStrategy appender.rolling.strategy.max 10 # 关闭 Netty channel handler 中不相关的错误警告 logger.netty.name org.jboss.netty.channel.DefaultChannelPipelinelogger.netty.level OFF jobmanager-service.yaml这是一个可选的Service资源仅在非HA模式下需要使用其主要作用是定义JobManager服务用于集群内部组件之间的通信。通过该ServiceJobManager的RPC端口用于任务调度通信、Blob Server端口用于管理二进制大对象和Web UI端口用于用户监控和管理界面访问得以暴露。具体配置如下 apiVersion: v1 kind: Service metadata:name: flink-jobmanager spec:type: ClusterIPports:- name: rpcport: 6123- name: blob-serverport: 6124- name: webuiport: 8081selector:app: flinkcomponent: jobmanagerjobmanager-rest-service.yaml同样是可选的Service该服务的作用是将JobManager的REST端口暴露为Kubernetes节点端口使得外部客户端可以通过节点的IP地址和指定端口访问Flink的Web UI界面方便用户对Flink作业进行监控、管理和操作。配置示例如下 apiVersion: v1 kind: Service metadata:name: flink-jobmanager-rest spec:type: NodePortports:- name: restport: 8081targetPort: 8081nodePort: 30081selector:app: flinkcomponent: jobmanagertaskmanager-query-state-service.yaml此Service也是可选的它的功能是将TaskManager的查询状态服务端口暴露为公共Kubernetes node的节点端口通过该端口可以访问Flink的Queryable State服务用于查询和管理TaskManager中的状态数据。配置内容如下 apiVersion: v1 kind: Service metadata:name: flink-taskmanager-query-state spec:type: NodePortports:- name: query-stateport: 6125targetPort: 6125nodePort: 30025selector:app: flinkcomponent: taskmanager当完成上述通用集群资源的配置文件编写后可以使用以下命令来创建这些服务资源

在执行以下指令时优先检查是否已经定义了通用集群资源声明

Configuration and service definition

kubectl create -f flink-configuration-configmap.yaml -n flink# service kubectl create -f jobmanager-service.yaml -n flink kubectl create -f jobmanager-rest-service.yaml -n flink kubectl create -f taskmanager-query-state-service.yaml -n flink2.3 Application模式部署 Flink Application集群是一种专门为运行单个Application而设计的专用集群在这种模式下部署集群时必须确保对应的Application能够正常运行。因此在提交Flink作业任务时首先需要启动Flink Application集群保证其处于可用状态然后才能进行作业提交操作。 在Kubernetes上部署一个基本的Flink Application集群时通常需要包含以下三个核心组件 一个运行JobManager的ApplicationJobManager在Flink集群中扮演着“大脑”的角色负责作业的调度、任务分配以及与TaskManager之间的协调通信确保整个作业的顺利执行。运行若干个TaskManager的DeploymentTaskManager是实际执行任务的工作节点它们接收来自JobManager的任务指令进行数据处理和计算工作。通过Deployment来管理TaskManager可以方便地实现任务节点的扩缩容以适应不同作业负载的需求。暴露JobManager上REST和UI端口的Service该Service的作用是将JobManager的REST API端口和Web UI端口暴露出来方便用户通过浏览器访问Flink的Web界面对作业进行监控、提交、暂停、取消等操作同时也便于外部系统通过REST API与Flink集群进行交互。 在进行Application集群资源定义时需要特别注意以下几点 jobmanager-job.yaml在该文件中args属性必须指定用户作业的主类这是告诉Flink集群从哪个类开始执行用户提交的作业。此外还可以参考如何设置JobManager参数的相关文档了解如何将额外的参数传递给jobmanager-job.yaml配置中指定的Flink镜像以满足不同作业的个性化需求。job artifacts参数该参数所对应的作业资源如Jar包等必须能够从资源定义示例中的job-artifacts-volume处获取。如果是在minikube集群中创建这些组件那么可以将定义示例中的job-artifacts-volume挂载为主机的本地目录方便快捷地提供作业资源。若不使用minikube集群也可以使用Kubernetes集群中任何其它可用类型的volume如PersistentVolumeClaim等来提供作业所需的资源。另外还可以选择构建一个已经包含job artifacts参数的自定义镜像将作业资源直接打包在镜像中简化资源管理流程。 以下是Apllication Mode集群资源定义模板在实际生产环境中使用时需要根据具体需求进行相应的调整 jobmanager-application-non-ha.yaml非HA模式下JobManager的配置模板示例 实在抱歉给你带来不好的体验我将分多次输出完整内容每次保证段落完整连贯。以下从Flink Standalone Kubernetes部署模式的Session模式继续。 2.4 Session模式部署 Flink Session集群以长期运行的Deployment形式存在可在其上运行多个作业。部署包含运行JobManager的Deployment、运行TaskManager的Deployment和暴露JobManager端口的Service。 资源定义模板 jobmanager-session-deployment-non-ha.yaml非HA模式下JobManager的Deployment配置。 apiVersion: apps/v1 kind: Deployment metadata:name: flink-jobmanager spec:replicas: 1selector:matchLabels:app: flinkcomponent: jobmanagertemplate:metadata:labels:app: flinkcomponent: jobmanagerspec:containers:- name: jobmanagerimage: apache/flink:1.14.4-scala_2.11args: [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: 9999volumes:- name: flink-config-volumeconfigMap:name: flink-configitems:- key: flink-conf.yamlpath: flink-conf.yaml- key: log4j-console.propertiespath: log4j-console.propertiesjobmanager-session-deployment-ha.yamlHA模式下JobManager的Deployment配置。 apiVersion: apps/v1 kind: Deployment metadata:name: flink-jobmanager spec:replicas: 1 # 通过设置大于 1 的整型值来开启 Standby JobManagerselector:matchLabels:app: flinkcomponent: jobmanagertemplate:metadata:labels:app: flinkcomponent: jobmanagerspec:containers:- name: jobmanagerimage: apache/flink:1.14.4-scala_2.11env:- name: POD_IPvalueFrom:fieldRef:apiVersion: v1fieldPath: status.podIP# 下面的 args 参数会使用 POD_IP 对应的值覆盖 config map 中 jobmanager.rpc.address 的属性值。args: [jobmanager, $(POD_IP)]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.propertiestaskmanager-session-deployment.yamlTaskManager的Deployment配置。 apiVersion: apps/v1 kind: Deployment metadata:name: flink-taskmanager spec:replicas: 2selector:matchLabels:app: flinkcomponent: taskmanagertemplate:metadata:labels:app: flinkcomponent: taskmanagerspec:containers:- name: taskmanagerimage: apache/flink:1.14.4-scala_2.11args: [taskmanager]ports:- containerPort: 6122name: rpc- containerPort: 6125name: query-statelivenessProbe:tcpSocket:port: 6122initialDelaySeconds: 30periodSeconds: 60volumeMounts:- name: flink-config-volumemountPath: /opt/flink/conf/securityContext: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集群操作命令 创建Session集群 kubectl create -f jobmanager-session-deployment.yaml kubectl create -f taskmanager-session-deployment.yaml停止Flink session集群 kubectl delete -f taskmanager-session-deployment.yaml kubectl delete -f jobmanager-session-deployment.yaml三、Flink Native Kubernetes部署模式 Flink Native kubernetes模式默认只启动jobmanager之后根据job任务提交情况动态的申请、启动taskmanager计算资源目前该模式支持session、application部署方式不支持per - job方式。 3.1 Flink Native KubernetesSession Flink Session Native on kubernetes和Flink 流程大致相同都需要构建基础dockerfile首先需要将获取的基础镜像push到本地仓库中其次才是构建镜像仓库。 编写dockerfile cat kubernete-native-session EOF

将以下内容填充到该docker文件

FROM apache/flink:1.14.6-scala_2.12 RUN rm -f /etc/localtime ln -sv /usr/share/zoneinfo/Asia/Shanghai /etc/localtime echo Asia/Shanghai /etc/timezone RUN export LANGzh_CN.UTF-8 EOF构建docker镜像 docker build -t ruijie/bigdata/flink-session:1.14.6-scala_2.12. –no-cache #推送到私有仓库中 docker push ruijie/bigdata/flink-session:1.14.6-scala_2.12创建flink native session kubernetes集群 ./bin/kubernetes-session.sh -Dkubernetes.cluster-idflink-native-cluster -Dkubernetes.container.imageruijie/bigdata/flink-session:1.14.6-scala_2.12 -Dkubernetes.namespaceflink -Dkubernetes.jobmanager.service-accountflink-service-account -Dkubernetes.rest-service.exposed.typeNodePort执行完毕之后可以得到以下结果我们指定的是nodePort而非clusterId,后续会对这一部分进行详细解释执行完毕之后我们的kubernetes 的flink native session创建完毕了可以通过日志打印看出web - ui暴露的地址进行访问。 通过访问控制台打印的日志可以找到web - ui访问地址或者通过kubectl get pods -n flink查看然后通过kubectl logs -f查看日志。 4. 提交任务 ./bin/flink run --target kubernetes-session -Dkubernetes.cluster-idflink-native-session -Dkubernetes.namespaceflink -Dkubernetes.jobmanager.service-accountflink-service-account ./examples/streaming/TopSpeedWindowing.jar ./examples/streaming/WordCount.jar -Dkubernetes.taskmanager.cpu2000m -Dexternal-resource.limits.kubernetes.cpu4000m -Dexternal-resource.limits.kubernetes.memory10Gi -Dexternal-resource.requests.kubernetes.cpu2000m -Dexternal-resource.requests.kubernetes.memory8Gi -Dkubernetes.taskmanager.cpu2000m 删除集群资源 kubectl delete deployment/flink-native-session取消正在运行的任务 echostop | ./bin/kubernetes-session.sh -Dkubernetes.cluster-idmy-first-flink-cluster -Dexecution.attachedtrue3.2 Flink Application Native Kubernetes Flink application on native 和以上的相同都需要经过dockerfile 文件的编写和构建镜像。 编写dockerfile文件 cat flink-applicationEOF FROM flink:1.14.6-scala_2.12 ENV FLINK_HOME/opt/flink RUN rm -f /etc/localtime ln -sv /usr/share/zoneinfo/Asia/Shanghai /etc/localtime echo Asia/Shanghai /etc/timezone RUN export LANGzh_CN.UTF-8 RUN mkdir -p \(FLINK_HOME/usrlib COPY TopSpeedWindowing.jar \)FLINK_HOME/usrlib/ EOF构建镜像与操作

构建镜像

docker build -f flink-application -t docker.ruijie.com/bigdata/application-flink:1.14.6. –no-cache

推动到本地仓库中

docker push docker.ruijie.com/bigdata/application-flink:1.14.6

删除本地镜像

docker rmi docker.ruijie.com/bigdata/application-flink:1.14.6

删除kubernetes镜像

crictl rmi docker.ruijie.com/bigdata/application-flink:1.14.6同时也需要构建和下发到其他node节点将镜像加载进去避免因找不到镜像而报错。命名空间和权限也相同参考以上步骤。 3. 启动任务 ./bin/flink run-application --target kubernetes-application -Dkubernetes.cluster-idflink-application-cluster -Dkubernetes.container.imagedocker.ruijie.com/bigdata/application/flink:1.14.6 -Dkubernetes.jobmanager.replicas1 -Dkubernetes.namespaceflink -Dkubernetes.jobmanager.service-accountflink-service-account -Dexternal-resource.limits.kubernetes.cpu2000m -Dkubernetes.websocket.timeout60000 -Dexternal-resource.limits.kubernetes.memory1Gi -Dexternal-resource.requests.kubernetes.cpu1000m -Dexternal-resource.requests.kubernetes.memory1Gi -Dkubernetes.rest-service.exposed.typeNodePort \local:///usrlib/TopSpeedWindowing.jar查看任务运行情况

查看pods和svc

kubectl get pods,svc -n flink

查看日志

kubectl logs -f rj-flink-cluster-f4d9b796-lqg7q -n flinknative application集群管理

列出集群上正在运行的作业

./bin/flink list –target kubernetes-application -Dkubernetes.cluster-idflink-native-session

取消正在运行的作业

./bin/flink cancel –target kubernetes-application -Dkubernetes.cluster-idflink-native-session jobId以上就是Flink与Kubernetes集成的全流程及各模式详细操作。如果在实际操作中有任何疑问或想了解某部分的更多细节欢迎随时告诉我。