文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

基于k8s如何部署Session模式Flink集群

2023-07-05 12:25

关注

这篇文章主要介绍“基于k8s如何部署Session模式Flink集群”,在日常操作中,相信很多人在基于k8s如何部署Session模式Flink集群问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”基于k8s如何部署Session模式Flink集群”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

基于k8s部署Session模式Flink集群

在分布式计算领域中,Apache Flink是一个快速、可靠且易于使用的计算引擎。Flink集群是一个分布式系统,它由Flink JobManager和多个Flink TaskManager组成。部署Flink集群时,高可用性是非常重要的一个考虑因素。

什么是Session模式

在Flink中,有两种部署模式:Standalone和Session。Standalone模式下,Flink集群是一组独立的进程,它们共享同一个配置文件,并通过Akka通信。Session模式下,Flink集群是动态的、可伸缩的,可以根据需要启动或停止。Session模式下,Flink JobManager和TaskManager进程运行在容器中,可以通过k8s进行动态管理。

Session模式的优点是:

因此,Session模式是在Kubernetes上部署Flink集群的首选模式。

Flink的filesystem

在 Flink 的处理过程中,数据可能会存储在不同的文件系统中,如本地文件系统、HDFS、S3 等。为了统一处理这些文件系统,Flink 引入了 FileSystem 的概念,它是一个抽象的接口,提供了对不同文件系统的统一访问方式。

fileSystem 的实现类可以通过 Flink 的配置文件指定。Flink 支持多种文件系统,包括本地文件系统、HDFS、S3、Google Cloud Storage 等,因为minio实现了s3协议,所以也可以使用minio来作为文件系统。

基于k8s部署高可用Session模式Flink集群

各组件版本号

组件版本号
kubernetes1.15.12
flink1.15.3

制作镜像

使用minio作为文件系统需要增加s3相关的依赖jar包,所以需要自己制作镜像

Dockerfile:

FROM apache/flink:1.15.3-scala_2.12# 需要用到的jar包# flink-cdcADD lib/flink-sql-connector-mysql-cdc-2.3.0.jar /opt/flink/lib/# jdbc连接器ADD lib/flink-connector-jdbc-1.15.3.jar /opt/flink/lib/# mysql驱动ADD lib/mysql-connector-j-8.0.32.jar /opt/flink/lib/# oracle驱动ADD lib/ojdbc8-21.9.0.0.jar /opt/flink/lib/# 文件系统插件需要放到插件目录,按规范放置RUN mkdir /opt/flink/plugins/s3-fs-presto && cp -f /opt/flink/opt/flink-s3-fs-presto-1.15.3.jar /opt/flink/plugins/s3-fs-presto/

构建镜像:

docker build -t sivdead/flink:1.15.3_scala_2.12 -f .\DockerFile .

配置文件(ConfigMap)

配置文件分两个部分,flink-conf.yamllog4j-console.properties

apiVersion: v1kind: ConfigMapmetadata:  name: flink-config  namespace: szyx-flink  labels:    app: flinkdata:  flink-conf.yaml: |+    kubernetes.cluster-id: szyx-flink    # 所在的命名空间    kubernetes.namespace: szyx-flink    jobmanager.rpc.address: flink-jobmanager    taskmanager.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: 2867m    parallelism.default: 2    execution.checkpointing.interval: 10s        # 文件系统    fs.default-scheme: s3    # minio地址    s3.endpoint: https://minio.k8s.io:9000    # minio的bucket    s3.flink.bucket: szyxflink    s3.access-key: <minio账号>    s3.secret-key: <minio密码>    # 状态存储格式    state.backend: rocksdb    s3.path.style.access: true    blob.storage.directory: /opt/flink/tmp/blob    web.upload.dir: /opt/flink/tmp/upload    io.tmp.dirs: /opt/flink/tmp    # 状态管理    # checkpoint存储地址    state.checkpoints.dir: s3://szyxflink/state/checkpoint    # savepoint存储地址    state.savepoints.dir: s3://szyxflink/state/savepoint    # checkpoint间隔    execution.checkpointing.interval: 5000    execution.checkpointing.mode: EXACTLY_ONCE    # checkpoint保留数量    state.checkpoints.num-retained: 3    # history-server# 监视以下目录中已完成的作业    jobmanager.archive.fs.dir: s3://szyxflink/completed-jobs    # 每 10 秒刷新一次    historyserver.archive.fs.refresh-interval: 10000    historyserver.archive.fs.dir: s3://szyxflink/completed-jobs    # 高可用    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory    high-availability.storageDir: s3://szyxflink/ha    # 每6个小时触发一次savepoint    kubernetes.operator.periodic.savepoint.interval: 6h    kubernetes.operator.savepoint.history.max.age: 24h    kubernetes.operator.savepoint.history.max.count: 5    # Restart of unhealthy job deployments    kubernetes.operator.cluster.health-check.enabled: true    # Restart failed job deployments     kubernetes.operator.job.restart.failed: true  log4j-console.properties: |+    # This affects logging for both user code and Flink    rootLogger.level = INFO    rootLogger.appenderRef.console.ref = ConsoleAppender    rootLogger.appenderRef.rolling.ref = RollingFileAppender    # Uncomment this if you want to _only_ change Flink's logging    #logger.flink.name = org.apache.flink    #logger.flink.level = INFO    # The following lines keep the log level of common libraries/connectors on    # log level INFO. The root logger does not override this. You have to manually    # change the log levels here.    logger.akka.name = akka    logger.akka.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    # Log all infos to the 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    # Log all infos in the given 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    # Suppress the irrelevant (wrong) warnings from the Netty channel handler    logger.netty.name = org.jboss.netty.channel.DefaultChannelPipeline    logger.netty.level = OFF

添加serviceAccount并授权

在 Kubernetes 上部署 Flink 集群时,需要创建一个 serviceAccount 来授权 Flink 任务在 Kubernetes 集群中执行。ServiceAccount 是 Kubernetes 中一种资源对象,用于授权 Pod 访问 Kubernetes API。当 Flink JobManager 或 TaskManager 启动时,需要使用这个 serviceAccount 来与 Kubernetes API 交互,获取集群资源并进行任务的调度和执行。

apiVersion: v1kind: ServiceAccountmetadata:  name: flink-service-account  namespace: szyx-flink---apiVersion: rbac.authorization.k8s.io/v1kind: Rolemetadata:  namespace: szyx-flink  name: flinkrules:- apiGroups: [""]  resources: ["pods", "services","configmaps"]  verbs: ["create", "get", "list", "watch", "delete"]- apiGroups: [""]  resources: ["pods/log"]  verbs: ["get"]- apiGroups: ["batch"]  resources: ["jobs"]  verbs: ["create", "get", "list", "watch", "delete"]- apiGroups: ["extensions"]  resources: ["ingresses"]  verbs: ["create", "get", "list", "watch", "delete"]---apiVersion: rbac.authorization.k8s.io/v1kind: RoleBindingmetadata:  namespace: szyx-flink  name: flink-role-bindingroleRef:  apiGroup: rbac.authorization.k8s.io  kind: Role  name: flinksubjects:- kind: ServiceAccount  name: flink-service-account  namespace: flink

部署JobManager

jobManager挂载用pvc

apiVersion: v1kind: PersistentVolumeClaimmetadata:  name: flink-tmp  namespace: szyx-flinkspec:  accessModes:    - ReadWriteOnce  resources:    requests:      storage: 40Gi

Deployment:

apiVersion: apps/v1kind: Deploymentmetadata:  name: flink-jobmanager  namespace: szyx-flinkspec:  replicas: 1 # Set the value to greater than 1 to start standby JobManagers  selector:    matchLabels:      app: flink      component: jobmanager  template:    metadata:      labels:        app: flink        component: jobmanager    spec:      containers:      - name: jobmanager        imagePullPolicy: Always        image: sivdead/flink:1.15.3_scala_2.12        env:        # 注入POD的ip到容器内        - name: POD_IP          valueFrom:            fieldRef:              apiVersion: v1              fieldPath: status.podIP        # 时区        - name: TZ          value: Asia/Shanghai        # The following args overwrite the value of jobmanager.rpc.address configured in the configuration config map to POD_IP.        args: ["jobmanager", "$(POD_IP)"]        ports:        - containerPort: 6123          name: rpc        - containerPort: 6124          name: blob-server        - containerPort: 8081          name: webui        livenessProbe:          tcpSocket:            port: 6123          initialDelaySeconds: 30          periodSeconds: 60        resources:          requests:            memory: "8192Mi"            cpu: "4"          limits:            memory: "8192Mi"            cpu: "4"        volumeMounts:        - name: flink-config-volume          mountPath: /opt/flink/conf        - name: tmp-dir          mountPath: /opt/flink/tmp        securityContext:          runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessary      serviceAccountName: flink-service-account # Service account which has the permissions to create, edit, delete ConfigMaps      # 节点选择器      nodeSelector:        zone: mainland      # 节点容忍      tolerations:        - key: zone          value: mainland          effect: NoSchedule      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: tmp-dir        persistentVolumeClaim:          claimName: flink-tmp

Service:

apiVersion: v1kind: Servicemetadata:  name: flink-jobmanagerspec:  type: ClusterIP  ports:  - name: rpc    port: 6123  - name: blob-server    port: 6124  - name: webui    port: 8081  selector:    app: flink    component: jobmanager

Ingress:

apiVersion: extensions/v1beta1kind: Ingressmetadata:  annotations:    # 因为有可能需要上传jar包,所以需要设置大一些    nginx.ingress.kubernetes.io/proxy-body-size: 300m    nginx.ingress.kubernetes.io/rewrite-target: /$1  name: job-manager  namespace: szyx-flinkspec:  rules:  - host: flink.k8s.io    http:      paths:      - backend:          serviceName: flink-jobmanager          servicePort: 8081        path: /flink/(.*)

访问http://flink.k8s.io/flink/能打开flink界面,说明部署完成

基于k8s如何部署Session模式Flink集群

部署TaskManager

Deployment:

apiVersion: apps/v1kind: Deploymentmetadata:  name: flink-taskmanager  namespace: szyx-flinkspec:  replicas: 2  selector:    matchLabels:      app: flink      component: taskmanager  template:    metadata:      labels:        app: flink        component: taskmanager    spec:      containers:      - name: taskmanager        imagePullPolicy: Always        image: sivdead/flink:1.15.3_scala_2.12        args: ["taskmanager"]        ports:        - containerPort: 6122          name: rpc        - containerPort: 6125          name: query-state        livenessProbe:          tcpSocket:            port: 6122          initialDelaySeconds: 30          periodSeconds: 60        volumeMounts:        - name: flink-config-volume          mountPath: /opt/flink/conf/        securityContext:          runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessary        resources:          requests:            memory: "8192Mi"            cpu: "4"          limits:            memory: "8192Mi"            cpu: "4"            # 节点选择器      nodeSelector:        zone: mainland      # 节点容忍      tolerations:        - key: zone          value: mainland          effect: NoSchedule      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

部署完成后,打开flink页面,查看TaskManages:

基于k8s如何部署Session模式Flink集群

测试提交作业

基于k8s如何部署Session模式Flink集群

运行作业

基于k8s如何部署Session模式Flink集群

检查运行结果

基于k8s如何部署Session模式Flink集群

基于k8s如何部署Session模式Flink集群

到此,关于“基于k8s如何部署Session模式Flink集群”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注编程网网站,小编会继续努力为大家带来更多实用的文章!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯