一. 环境

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>