一. 环境
1. 可用的k8s集群
2. 宿主机有flink
3. 创建k8s命名空间
# 创建namespace
kubectl create namespace flink-session-test
# 新建一个serviceaccount, 提交flink
kubectl create serviceaccount flink -n flink-session-test
# 绑定
kubectl create clusterrolebinding flink-role-binding-flink-session-test_flink \
--clusterrole=edit --serviceaccount=flink-session-test:flink
二. session模式
1. 创建session
其他参数可以官网查看:
https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/deployment/config/#kubernetes-rest-service-exposed-type
说明:session的方式是无法加载docker镜像的,-Dkubernetes.container.image这个参数不会生效
/opt/frame/flink-1.14.4/bin/kubernetes-session.sh \
-Dkubernetes.namespace=flink-session-test \
-Dkubernetes.jobmanager.service-account=flink \
-Dkubernetes.cluster-id=k8s-flink-application-test \
-Dtaskmanager.memory.process.size=1600m \
-Djobmanager.memory.process.size=1020m \
-Dkubernetes.taskmanager.cpu=1 \
-Dtaskmanager.numberOfTaskSlots=4 \
-Dkubernetes.rest-service.exposed.type=NodePort \
-Dresourcemanager.taskmanager-timeout=3600000
2. 提交job
/opt/frame/flink-1.14.4/bin/flink run -d \
-e kubernetes-session \
-Dkubernetes.namespace=flink-session-cluster-test-1213 \
-Dkubernetes.cluster-id=k8s-flink-application-cluster \
-c com.cti.service.stream.test.K8sConnectTest \
/home/opsadmin/flink/flink-stream-service-stg-1.0-SNAPSHOT-jar-with-dependencies.jar
二. application模式
1. 写Dockerfile
需要把运行任务的jar包也放到镜像中
FROM flink:1.13.1-scala_2.12-java8
RUN mkdir -p /opt/flink/usrlib
COPY ./flink-conf.yaml /opt/flink/conf/flink-conf.yaml
COPY ./flink-stream-service-stg-1.0-SNAPSHOT-jar-with-dependencies.jar /opt/flink/usrlib/flink-stream-service-stg-1.0-SNAPSHOT-jar-with-dependencies.jar
2. 构建镜像
docker build -t docker.flink:v1.13.1-2.12-1.8 .
3. 启动任务
/opt/frame/flink-1.14.4/bin/flink run-application \
--target kubernetes-application \
-Dkubernetes.namespace=flink-session-cluster-test-1213 \
-Dkubernetes.jobmanager.service-account=flink \
-Dkubernetes.cluster-id=my-first-application-cluster \
-Dkubernetes.container.image=docker.flink:v1.13.1-2.12-1.8 \
-Dkubernetes.rest-service.exposed.type=NodePort \
-c com.cti.service.stream.test.K8sConnectTest \
local:///opt/flink/usrlib/flink-stream-service-stg-1.0-SNAPSHOT-jar-with-dependencies.jar
# 说明 local:// 后面接的是容器中的jar包位置
Create flink application cluster my-first-application-cluster successfully, JobManager Web Interface: http://xxxx:31249
可以打开网址查看你的任务
4. 报错查看日志
kubectl get pods --all-namespaces -o wide
# 查看具体报错原因
kubectl logs --tail 200 -n flink-session-cluster-test-1213 -f my-first-application-cluster-taskmanager-1-1
5. 查看停止任务
# 查看
/opt/frame/flink-1.14.4/bin/flink list --target kubernetes-application -Dkubernetes.cluster-id=my-first-application-cluster -Dkubernetes.namespace=flink-session-cluster-test-1213
# 停止
/opt/frame/flink-1.14.4/bin/flink cancel --target kubernetes-application -Dkubernetes.cluster-id=my-first-application-cluster -Dkubernetes.namespace=flink-session-cluster-test-1213 <jobId>
