Flink Kubernates Native - 入门-编程知识

创建 namespace

[root@CentOSA flink-1.17.1]# kubectl create ns flink-native
[root@CentOSA flink-1.17.1]# kubectl config set-context --current --namespace=flink-native

命令空间添加资源限制

[root@CentOSA flink-1.17.1]# vim namespace-ResourceQuota.yaml

apiVersion: v1 
kind: ResourceQuota
metadata:name: mem-cpu-quotanamespace: flink-native
spec:hard:requests.cpu: "10"requests.memory: 10Gilimits.cpu: "10"limits.memory: 10Gi

[root@CentOSA flink-1.17.1]# kubectl apply -f namespace-ResourceQuota.yaml

创建用户并赋权

[root@CentOSA flink-1.17.1]# kubectl create serviceaccount ric
[root@CentOSA flink-1.17.1]# kubectl create clusterrolebinding flink-role-binding-flink --clusterrole=edit --serviceaccount=flink-native:ric --namespace=flink-native

提前在所有的flnk节点拉取flink镜像

[root@CentOSA flink-1.17.1]# docker pull apache/flink:1.17.1-scala_2.12 
[root@CentOSA flink-1.17.1]# docker pull flink:1.17.1-scala_2.12

SessionMode

创建Flink 会话

./bin/kubernetes-session.sh \-Dkubernetes.namespace=flink-native \-Dkubernetes.jobmanager.service-account=ric \-Dkubernetes.cluster-id=session001 \-Dtaskmanager.memory.process.size=1024m \-Dkubernetes.taskmanager.cpu=1 \-Dtaskmanager.numberOfTaskSlots=4 \-Dresourcemanager.taskmanager-timeout=3600000 \-Dkubernetes.rest-service.exposed.type=NodePort \-Dkubernetes.container.image.ref=flink:1.17.1-scala_2.12

2023-08-28 15:50:06,802 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: taskmanager.memory.process.size, 1728m
2023-08-28 15:50:06,805 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.bind-host, localhost
2023-08-28 15:50:06,805 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: taskmanager.bind-host, localhost
2023-08-28 15:50:06,805 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: taskmanager.host, localhost
2023-08-28 15:50:06,805 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: parallelism.default, 1
2023-08-28 15:50:06,805 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.execution.failover-strategy, region
2023-08-28 15:50:06,805 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.rpc.address, localhost
2023-08-28 15:50:06,806 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2023-08-28 15:50:06,806 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: rest.address, localhost
2023-08-28 15:50:06,806 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.memory.process.size, 1600m
2023-08-28 15:50:06,806 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.rpc.port, 6123
2023-08-28 15:50:06,806 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: rest.bind-address, localhost
2023-08-28 15:50:06,900 INFO  org.apache.flink.client.deployment.DefaultClusterClientServiceLoader [] - Could not load factory due to missing dependencies.
2023-08-28 15:50:08,271 INFO  org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils [] - The derived from fraction jvm overhead memory (160.000mb (167772162 bytes)) is less than its min value 192.000mb (201326592 bytes), min value will be used instead
2023-08-28 15:50:08,291 INFO  org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils [] - The derived from fraction jvm overhead memory (102.400mb (107374184 bytes)) is less than its min value 192.000mb (201326592 bytes), min value will be used instead
2023-08-28 15:50:08,292 INFO  org.apache.flink.runtime.util.config.memory.ProcessMemoryUtils [] - The derived from fraction network memory (57.600mb (60397978 bytes)) is less than its min value 64.000mb (67108864 bytes), min value will be used instead
2023-08-28 15:50:08,498 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration blob.server.port will be set to 6124
2023-08-28 15:50:08,498 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration taskmanager.rpc.port will be set to 6122
2023-08-28 15:50:09,989 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Create flink session cluster session001 successfully, JobManager Web Interface: http://192.168.42.130:32432

可以根据提示信息访问JobManager Web Interface: http://192.168.42.130:32432页面
Session会话页面

提交任务到指定会话

运行flink session任务

 ./bin/flink run \--target kubernetes-session \-Dkubernetes.cluster-id=session001 \-Dkubernetes.namespace=flink-native \-Dkubernetes.jobmanager.service-account=ric \-Dkubernetes.rest-service.exposed.type=NodePort \./examples/streaming/TopSpeedWindowing.jar

Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
2023-08-28 15:54:02,974 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Retrieve flink cluster session001 successfully, JobManager Web Interface: http://192.168.42.130:32432
Job has been submitted with JobID 465dd11e73b5c9ac49c8e2945bf862b2

提交任务结果

查看任务列表

 ./bin/flink list \--target kubernetes-session \-Dkubernetes.cluster-id=session001 \-Dkubernetes.namespace=flink-native \-Dkubernetes.jobmanager.service-account=ric \-Dkubernetes.rest-service.exposed.type=NodePort \

2023-08-28 15:57:33,654 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Retrieve flink cluster session001 successfully, JobManager Web Interface: http://192.168.42.130:32432
Waiting for response...
------------------ Running/Restarting Jobs -------------------
28.08.2023 15:54:03 : 465dd11e73b5c9ac49c8e2945bf862b2 : CarTopSpeedWindowingExample (RUNNING)
--------------------------------------------------------------
No scheduled jobs.

取消任务：

 ./bin/flink cancel \--target kubernetes-session \-Dkubernetes.cluster-id=session002 \-Dkubernetes.namespace=flink-native \-Dkubernetes.jobmanager.service-account=ric \-Dkubernetes.rest-service.exposed.type=NodePort \465dd11e73b5c9ac49c8e2945bf862b2

Cancelling job 465dd11e73b5c9ac49c8e2945bf862b2.
2023-08-28 16:02:43,209 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Retrieve flink cluster session001 successfully, JobManager Web Interface: http://192.168.42.130:32432
Cancelled job 465dd11e73b5c9ac49c8e2945bf862b2.

删除session作业：

[root@CentOSA flink-1.17.1]# kubectl delete deployment session001 -n flink-native
deployment.apps "session001" deleted

Application Model

构建docker镜像

1、编写DockerFile

[root@CentOSA flink-1.17.1] cd /export/server/app-mode
[root@CentOSA app-mode] vi Dockerfile
FROM flink
RUN mkdir -p $FLINK_HOME/usrlib
COPY TopSpeedWindowing.jar $FLINK_HOME/usrlib/TopSpeedWindowing.jar

2、构建镜像文件

[root@CentOSC app-mode]# docker build -t my-flink-job:1.0 ./
[+] Building 0.6s (8/8) FINISHED                                                                                                                                                                                      docker:default=> [internal] load .dockerignore                                                                                                                                                                                               0.1s=> => transferring context: 2B                                                                                                                                                                                                 0.0s=> [internal] load build definition from Dockerfile                                                                                                                                                                            0.1s=> => transferring dockerfile: 271B                                                                                                                                                                                            0.0s=> [internal] load metadata for docker.io/library/flink:1.17.1-scala_2.12                                                                                                                                                      0.0s=> [1/3] FROM docker.io/library/flink:1.17.1-scala_2.12                                                                                                                                                                        0.1s=> [internal] load build context                                                                                                                                                                                               0.0s=> => transferring context: 17.04kB                                                                                                                                                                                            0.0s=> [2/3] RUN mkdir -p /opt/flink/usrlib                                                                                                                                                                                        0.3s=> [3/3] COPY TopSpeedWindowing.jar  /opt/flink/usrlib/TopSpeedWindowing.jar                                                                                                                                                   0.0s=> exporting to image                                                                                                                                                                                                          0.0s=> => exporting layers                                                                                                                                                                                                         0.0s=> => writing image sha256:a017db16aaa8086045dc1b28315e6e2fe47113aa41e1a0f547b58c1b191dac4a                                                                                                                                    0.0s=> => naming to docker.io/library/my-flink-job:1.0                                                                                                                                                                             0.0s
[root@CentOSC appMode]# docker images
REPOSITORY                                           TAG                 IMAGE ID       CREATED              SIZE
my-flink-job                                         1.0                 a017db16aaa8   About a minute ago   782MB
flannel/flannel                                      v0.22.2             d73868a08083   10 days ago          70.2MB
flink                                                1.17.1-scala_2.12   efd18635797b   11 days ago          782MB
flannel/flannel-cni-plugin                           v1.2.0              a55d1bad692b   5 weeks ago          8.04MB
apache/flink                                         1.17.1-scala_2.12   fef5f770f8b3   3 months ago         780MB
registry.aliyuncs.com/google_containers/kube-proxy   v1.20.5             5384b1650507   2 years ago          118MB
registry.aliyuncs.com/google_containers/pause        3.2                 80d28bedfe5d   3 years ago          683kB

启动任务

 ./bin/flink run-application \--target kubernetes-application \-Dkubernetes.cluster-id=my-first-application-cluster \-Dkubernetes.container.image=my-flink-job:1.0 \-Dkubernetes.namespace=flink-native \-Dkubernetes.jobmanager.service-account=ric \-Dkubernetes.rest-service.exposed.type=NodePort \

2023-08-28 16:07:45,180 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration blob.server.port will be set to 6124
2023-08-28 16:07:45,181 INFO  org.apache.flink.kubernetes.utils.KubernetesUtils            [] - Kubernetes deployment requires a fixed port. Configuration taskmanager.rpc.port will be set to 6122
2023-08-28 16:07:46,742 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Create flink application cluster my-first-application-cluster successfully, JobManager Web Interface: http://192.168.42.130:32003

App 应用模型

查看任务列表

./bin/flink list \--target kubernetes-application \-Dkubernetes.cluster-id=my-first-application-cluster \-Dkubernetes.namespace=flink-native \-Dkubernetes.jobmanager.service-account=ric \-Dkubernetes.rest-service.exposed.type=NodePort

2023-08-28 16:10:00,093 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Retrieve flink cluster my-first-application-cluster successfully, JobManager Web Interface: http://192.168.42.130:32003
Waiting for response...
------------------ Running/Restarting Jobs -------------------
28.08.2023 16:07:59 : c9ec57a2003a808ef5df04219fc7aacb : CarTopSpeedWindowingExample (RUNNING)
--------------------------------------------------------------
No scheduled jobs.

删除任务

 ./bin/flink cancel \--target kubernetes-application \-Dkubernetes.cluster-id=my-first-application-cluster \-Dkubernetes.namespace=flink-native \-Dkubernetes.jobmanager.service-account=ric \-Dkubernetes.rest-service.exposed.type=NodePort \c9ec57a2003a808ef5df04219fc7aacb

2023-08-28 16:11:16,963 INFO  org.apache.flink.kubernetes.KubernetesClusterDescriptor      [] - Retrieve flink cluster my-first-application-cluster successfully, JobManager Web Interface: http://192.168.42.130:32003
Cancelled job c9ec57a2003a808ef5df04219fc7aacb.