devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

【Flink系列】部署篇（三）：Native k8s部署高可用Flink集群实战_flink native和operator的区别、

作者：从前慢现在也慢 | 2024-06-13 21:24:03

赞

踩

flink native和operator的区别、

上一篇博文介绍了如何独立部署一个高可用的Flink集群,本篇介绍如何用Native k8s去部署高可用的Flink 集群。本篇介绍的集群构建在AWS上，和构建在自己的服务器相比，主要区别在文件系统的选择和使用上。我选用的S3服务。

  • EC2操作系统：centos7
  • 本机操作系统：Mac
  • flink version: 1.14
  • jdk version: java8
  • HA service: k8s
  • File System: S3

启动EC2

在AWS上启动3个EC2,操作系统选centos7,注意每个EC2要关联弹性ip地址，允许外网访问。
可参考https://wiki.centos.org/Cloud/AWS

部署k8s集群

安装docker

sudo yum-config-manager --add-repo https://download.docker.com/linux/centos/docker-ce.repo
sudo yum install docker -y
sudo systemctl start docker.service
sudo systemctl enable docker.service
sudo systemctl status docker

  • 1
  • 2
  • 3
  • 4
  • 5

安装k8s

$ sudo vi /etc/yum.repos.d/kubernetes.repo
[kubernetes]
name=Kubernetes
baseurl=https://packages.cloud.google.com/yum/repos/kubernetes-el7-x86_64
enabled=1
gpgcheck=1
repo_gpgcheck=0
gpgkey=https://packages.cloud.google.com/yum/doc/yum-key.gpg https://packages.cloud.google.com/yum/doc/rpm-package-key.gpg
$ sudo yum install -y kubelet
$ sudo yum install -y kubeadm
$ sudo systemctl enable kubelet

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

Linux环境准备

  1. set hostnames
// On your **Master** node, update your hostname using the following command:

//Master:192.168.100.90 (3.101.77.138)

$ sudo hostnamectl set-hostname master-node

//worker-node1:192.168.100.29 (3.101.77.139)

$ sudo hostnamectl set-hostname worker-node1

// worker-node2: 192.168.100.21 (3.101.77.140)

$ sudo hostnamectl set-hostname worker-node2

// Make a host entry or DNS record to resolve the hostname for all nodes:

$ sudo vi /etc/hosts

// With the entry:

192.168.100.63  master-node

192.168.100.36 node1 worker-node1

192.168.100.125 node2 worker-node2

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  1. 关闭SElinux
$ sudo setenforce 0
$ sudo sed -i --follow-symlinks 's/SELINUX=enforcing/SELINUX=disabled/g' /etc/sysconfig/selinux
$ sudo reboot
$ sestatus
SELinux status:                 disabled

  • 1
  • 2
  • 3
  • 4
  • 5
  1. update iptables config
$ sudo vi /etc/sysctl.d/k8s.conf
net.bridge.bridge-nf-call-ip6tables = 1
net.bridge.bridge-nf-call-iptables = 1
$ sudo sysctl --system

  • 1
  • 2
  • 3
  • 4
  1. disable swap(3)
$ sudo sed -i '/swap/d' /etc/fstab
$ sudo swapoff -a

  • 1
  • 2

部署k8s集群

// only master node
$ sudo kubeadm init
kubeadm join 192.168.100.63:6443 --token snmpyy.b4y506h6hr9u7fxh \
        --discovery-token-ca-cert-hash sha256:87a099765ce369c519bc02af84a6d4732b1cb987d3e95277b334e3cfc3aa0960
// Create required directories and start managing Kubernetes cluster
$ mkdir -p $HOME/.kube
$ sudo cp -i /etc/kubernetes/admin.conf $HOME/.kube/config
$ sudo chown $(id -u):$(id -g) $HOME/.kube/config
[ec2-user@master-node kubernetes]$ kubectl get nodes
NAME          STATUS     ROLES           AGE   VERSION
master-node   NotReady   control-plane   84m   v1.24.2
//Set up Pod network for the Cluster
$ export kubever=$(kubectl version | base64 | tr -d '\n')
$ kubectl apply -f https://cloud.weave.works/k8s/net?k8s-version=$kubever
// add nodes to your cluster
//在两个工作节点操作
$ kubeadm join 192.168.100.63:6443 --token snmpyy.b4y506h6hr9u7fxh \
        --discovery-token-ca-cert-hash sha256:87a099765ce369c519bc02af84a6d4732b1cb987d3e95277b334e3cfc3aa0960
//在master节点
$ kubectl label node worker-node1 node-role.kubernetes.io/worker=worker
$ kubectl label node worker-node2 node-role.kubernetes.io/worker=worker


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

Session Mode 部署Flink集群

安装Java8

$ sudo yum search java|grep jdk
$ sudo yum install -y java-1.8.0-openjdk
$ sudo yum install java-1.8.0-openjdk-devel -y
$ java -version

  • 1
  • 2
  • 3
  • 4

下载解压Flink安装包

$ cd /opt
$ wget https://dlcdn.apache.org/flink/flink-1.15.2/flink-1.15.2-bin-scala_2.12.tgz --no-check-certificate
$ tar -xzf flink-*.tgz

  • 1
  • 2
  • 3

启动集群（无高可用服务版本）

Native k8s是把k8s植入到了Flink安装包中，直接使用Flink的命令就可以在k8s集群中启动flink组件相关的pod。

//service account with RBAC permissions to create, delete pods
$ kubectl create namespace flink-cluster
$ kubectl create serviceaccount flink -n flink-cluster
$ kubectl create clusterrolebinding flink-role-binding-flink \
  --clusterrole=edit \
  --serviceaccount=flink-cluster:flink
//启动session集群1（此时读取的配置为master节点的配置文件，与taskmanager节点的配置文件无关）
$ ./bin/kubernetes-session.sh -Dkubernetes.namespace=flink-cluster -Dkubernetes.jobmanager.service-account=flink -Dkubernetes.cluster-id=my-session -Dtaskmanager.numberOfTaskSlots=6  -Dkubernetes.rest-service.exposed.type=NodePort
2022-07-01 08:46:49,621 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Create flink session cluster my-session successfully, JobManager Web Interface: http://192.168.100.63:32172
// Dashboard: http://3.101.77.141:32172/,此时没有任何资源\

//只部署了jobmanager，部署在worker-node1,两个服务，一个对内，一个对外
[root@master-node ec2-user]# kubectl get pods,svc,ep -n flink-cluster -o wide
NAME                      TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)             AGE    SELECTOR
service/my-session        ClusterIP   None            <none>        6123/TCP,6124/TCP   124m   app=my-session,component=jobmanager,type=flink-native-kubernetes
service/my-session-rest   NodePort    10.109.225.42   <none>        8081:31595/TCP      124m   app=my-session,component=jobmanager,type=flink-native-kubernetes

NAME                              READY   STATUS    RESTARTS   AGE    IP          NODE           NOMINATED NODE   READINESS GATES
pod/my-session-556f44f44b-94gfk   1/1     Running   0          124m   10.44.0.2   worker-node1   <none>           <none>

NAME                        ENDPOINTS                       AGE
endpoints/my-session        10.44.0.2:6124,10.44.0.2:6123   124m
endpoints/my-session-rest   10.44.0.2:8081                  124m
[root@master-node ec2-user]# kubectl get deployment -o wide -n flink-cluster
NAME         READY   UP-TO-DATE   AVAILABLE   AGE    CONTAINERS             IMAGES                           SELECTOR
my-session   1/1     1            1           125m   flink-main-container   apache/flink:1.14.5-scala_2.12   app=my-session,component=jobmanager,type=flink-native-kubernetes


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27

可以通过命令行提交，也可以通过dashboard提交任务

// run之后才会显示资源
$ ./bin/flink run \
    --target kubernetes-session \
		-Dkubernetes.namespace=flink-cluster \
    -Dkubernetes.cluster-id=my-session \
    ./examples/streaming/TopSpeedWindowing.jar
$ ./bin/flink run     --target kubernetes-session     -Dkubernetes.namespace=flink-cluster     -Dkubernetes.cluster-id=my-session   -Dparallelism.default=2  ./examples/streaming/TopSpeedWindowing.jar
//自动扩容功能测试
//一个taskmanager的任务槽使用完之前，cluster只有一个task manager
NAME                          READY   STATUS    RESTARTS   AGE   IP          NODE           NOMINATED NODE   READINESS GATES
my-session-556f44f44b-zdvrf   1/1     Running   0          21m   10.36.0.1   worker-node2   <none>           <none>
my-session-taskmanager-1-1    1/1     Running   0          20m   10.44.0.1   worker-node1   <none>           <none>
//继续提交作业，cluster会自动扩容
NAME                          READY   STATUS    RESTARTS   AGE   IP          NODE           NOMINATED NODE   READINESS GATES
my-session-556f44f44b-zdvrf   1/1     Running   0          21m   10.36.0.1   worker-node2   <none>           <none>
my-session-taskmanager-1-1    1/1     Running   0          20m   10.44.0.1   worker-node1   <none>           <none>
my-session-taskmanager-1-2    1/1     Running   0          34s   10.36.0.2   worker-node2   <none>           <none>
//当两个taskmanager的资源都用完了
[root@master-node flink-1.14.5]kubectl get pods -n flink-cluster -o wide
NAME                         READY   STATUS    RESTARTS   AGE     IP          NODE           NOMINATED NODE   READINESS GATES
my-session-58bb97cdc-85ssq   1/1     Running   0          17m     10.36.0.1   worker-node2   <none>           <none>
my-session-taskmanager-1-1   1/1     Running   0          8m26s   10.44.0.2   worker-node1   <none>           <none>
my-session-taskmanager-1-2   1/1     Running   0          3m11s   10.36.0.3   worker-node2   <none>           <none>
my-session-taskmanager-1-3   0/1     Pending   0          33s     <none>      <none>         <none>           <none>
//清理资源
$ kubectl delete deployment/my-session -n flink-cluster
$ kubectl delete clusterrolebinding flink-role-binding-flink
$ kubectl delete serviceaccount flink -n flink-cluster
$ kubectl delete namespace flink-cluster

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

和独立部署相比，采用Native k8s部署有以下几个值得注意的点：

  1. JM 和TM 只能部署在K8s集群中的工作节点上，因此对于三个节点组成的k8s集群而言，Flink集群事实上只部署在其中两个节点上。因此对于小规模集群，我认为采用Native k8s部署有些浪费资源。
  2. 尽管机器为8核，但是如果直接将-Dtaskmanager.numberOfTaskSlots参数设为8，集群会启动失败，因为默认每个taskmanager会按照slots的数量来分配CPU。当前两个工作节点要部署一个JM，两个TM，导致JM必须和TM挤在同一个节点上，JM默认占1个CPU核，出了JM和TM外，还有一些系统进程也需要CPU，大概占0.1个CPU，因此我这里将-Dtaskmanager.numberOfTaskSlots设为6，经验证6.9也是ok的。 如果希望设置更大的slots数，可以修改kubernetes.taskmanager.cpu，让CPU数和slots数不再关联。
  3. 在独立部署中，不同的TM可以配置不同的内存，不同的slots数量。但在Native k8s部署中，这些参数是所有TM共享的，不可以分开配置。
  4. 从上面的操作过程可以看到，采用native k8s启动flink Session集群事实上仅仅启动了jobmanager。而taskmanager是在任务提交之后根据任务需要的资源数来启动的。在K8s集群资源的限制内，需要多少TM就启动多少TM对应的pod。任务结束对应的资源也会释放。资源的分配和调度都交给了k8s，相比手工设置，更加自动化。但是，如果我们不想要flink组件把k8s所有的节点都用完了，想要限定节点的范围改怎么办？下面是一个通过node-selector指定jobmanager的节点示例
//启动Session cluster2, 修改配置
$ kubectl label nodes worker-node2 node=master
[root@master-node flink-1.14.5]# kubectl get nodes -l "node=master"
NAME           STATUS   ROLES    AGE   VERSION
worker-node2   Ready    worker   10d   v1.24.2
$ ./bin/kubernetes-session.sh -Dkubernetes.namespace=flink-cluster -Dkubernetes.jobmanager.service-account=flink -Dkubernetes.cluster-id=my-session -Dtaskmanager.numberOfTaskSlots=8  -D
kubernetes.rest-service.exposed.type=NodePort -Dkubernetes.jobmanager.node-selector=node:master
[root@master-node flink-1.14.5]# kubectl get pods,svc,ep -n flink-cluster -o wide
NAME                             READY   STATUS    RESTARTS   AGE   IP          NODE           NOMINATED NODE   READINESS GATES
pod/my-session-58bb97cdc-85ssq   1/1     Running   0          57s   10.36.0.1   worker-node2   <none>           <none>

NAME                      TYPE        CLUSTER-IP    EXTERNAL-IP   PORT(S)             AGE   SELECTOR
service/my-session        ClusterIP   None          <none>        6123/TCP,6124/TCP   57s   app=my-session,component=jobmanager,type=flink-native-kubernetes
service/my-session-rest   NodePort    10.108.58.5   <none>        8081:31181/TCP      57s   app=my-session,component=jobmanager,type=flink-native-kubernetes

NAME                        ENDPOINTS                       AGE
endpoints/my-session        10.36.0.1:6124,10.36.0.1:6123   57s
endpoints/my-session-rest   10.36.0.1:8081                  57s

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18

启动集群（含高可用服务版本）

高可用服务的部署需要一个可共享的持久化存储目录，因为部署在AWS上，这里我选择S3。
因此首先要解决的问题是如何让集群可以使用S3

配置s3

  1. 首先要赋予EC2 访问S3 bucket的权限，这可以通过添加IAM实现。
  2. Copy the respective JAR file（flink-s3-fs-hadoop-1.14.4.jar） from the opt directory to the plugins directory of your Flink distribution before starting Flink. 直接cp即可
  3. 启动集群的时候enable内置的插件就可以使用s3了
./bin/kubernetes-session.sh
    -Dcontainerized.master.env.ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.14.4.jar \
    -Dcontainerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.14.4.jar

  • 1
  • 2
  • 3

Native k8s部署下使用s3的坑较多，各种报错。主要有两种：

  1. 没有s3访问的权限。检查IAM的配置，检查s3的访问控制。
  2. 找不到对应的类。这主要是插件没有使用对导致的。

启动高可用集群

./bin/kubernetes-session.sh -Dkubernetes.namespace=flink-cluster -Dkubernetes.jobmanager.service-account=flink -Dkubernetes.cluster-id=my-session -Dtaskmanager.numberOfTaskSlots=6  -Dkubernetes.rest-service.exposed.type=NodePort -Dcontainerized.master.env.ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.14.5.jar -Dcontainerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.14.5.jar -Dhigh-availability=org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory -Dhigh-availability.storageDir=s3a://yunzpeng-bucket/flink-ha 

  • 1

总结：Native k8s部署下session模式集群启动和作业提交过程

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xdKpmvz1-1673498615890)(2022-12-22-00-00-17.png)]

  • 第一个阶段：启动 Session Cluster。Flink Client 内置了 K8s Client，告诉 K8s Master 创建 Flink Master Deployment，ConfigMap，SVC。创建完成后，Master 就拉起来了。这时，Session 就部署完成了，并没有维护任何 TaskManager。
  • 第二个阶段：当用户提交 Job 时，可以通过 Flink Client 或者 Dashboard 的方式，然后通过 Service 到 Dispatcher，Dispatcher 会产生一个 JobMaster。JobMaster 会向 K8sResourceManager 申请资源。ResourceManager 会发现现在没有任何可用的资源，它就会继续向 K8s 的 Master 去请求资源，请求资源之后将其发送回去，起新的 Taskmanager。Taskmanager 起来之后，再注册回来，此时的 ResourceManager 再向它去申请 slot 提供给 JobMaster，最后由 JobMaster 将相应的 Task 部署到 TaskManager 上。这样整个从 Session 的拉起到用户提交都完成了。

Applicatiob Mode部署Flink 集群

无高可用

// 在root下操作
// 1. build a docker image with the flink job
FROM flink:1.14.5
RUN mkdir -p $FLINK_HOME/usrlib
COPY ./examples/streaming/TopSpeedWindowing.jar $FLINK_HOME/usrlib/my-flink-job.jar
$ docker build -t pandafish1996/flink-demo .
// 2. Push image to image warehouse
$ docker login --username=pandafish1996
// 规范： docker push 注册用户名/镜像名
//$ docker tag yunzpeng/flink-word-count pandafish1996/flink-word-count
$ docker push pandafish1996/flink-demo 

//3. start a flink application cluster
$ kubectl create namespace flink-cluster
$ kubectl create serviceaccount flink -n flink-cluster
$ kubectl create clusterrolebinding flink-role-binding-flink \
  --clusterrole=edit \
  --serviceaccount=flink-cluster:flink
// without HA
$ ./bin/flink run-application \
--target kubernetes-application \
-Dkubernetes.cluster-id=my-first-application-cluster \
-Dkubernetes.container.image=pandafish1996/flink-demo \
-Dkubernetes.namespace=flink-cluster \
-Dkubernetes.service-account=flink \
-Dparallelism.default=2 \
-Dtaskmanager.numberOfTaskSlots=6 \
-Dkubernetes.rest-service.exposed.type=NodePort  \
local:///opt/flink/usrlib/my-flink-job.jar

$ kubectl delete deployment/my-first-application-cluster -n flink-cluster

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31

高可用版本

// 1. 构造image并上传到远程仓库
FROM flink:1.14.5
RUN mkdir -p $FLINK_HOME/usrlib
COPY ./examples/streaming/TopSpeedWindowing.jar $FLINK_HOME/usrlib/my-flink-job.jar
RUN mkdir -p $FLINK_HOME/plugins/flink-s3-fs-hadoop
COPY ./opt/flink-s3-fs-hadoop-1.14.5.jar $FLINK_HOME/plugins/flink-s3-fs-hadoop/
$ docker build -t pandafish1996/flink-hademo .
$ docker login --username=pandafish1996
// 规范： docker push 注册用户名/镜像名
//$ docker tag yunzpeng/flink-word-count pandafish1996/flink-word-count
$ docker push pandafish1996/flink-hademo 
// 2. 启动集群
./bin/flink run-application \
--target kubernetes-application \
-Dkubernetes.cluster-id=ha-cluster1 \
-Dkubernetes.container.image=pandafish1996/flink-hademo \
-Dkubernetes.namespace=flink-cluster \
-Dkubernetes.service-account=flink \
-Dparallelism.default=2 \
-Dtaskmanager.numberOfTaskSlots=6 \
-Dkubernetes.rest-service.exposed.type=NodePort  \
-Dhigh-availability=org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory \
-Dhigh-availability.storageDir=s3://yunzpeng-bucket/flink-ha \
-Drestart-strategy=fixed-delay \
-Drestart-strategy.fixed-delay.attempts=10 \
-Dcontainerized.master.env.ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.14.5.jar \
-Dcontainerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS=flink-s3-fs-hadoop-1.14.5.jar \
local:///opt/flink/usrlib/my-flink-job.jar

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28

参考

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/714905
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号