flink on k8s提交任务
目录
- 相关文档
- 前置准备
- 构建镜像
- 提交任务
相关文档
https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/resource-providers/native_kubernetes/
前置准备
flink的lib目录下放入两个依赖
bcpkix-jdk15on-1.68.jar
bcprov-jdk15on-1.69.jar
创建用户有RABC权限去执行创建pods
官网原文：https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/resource-providers/native_kubernetes/#rbac
k8s中每个命名空间都有一个默认服务帐户。但是，default 服务帐户可能没有在 Kubernetes 集群中创建或删除 Pod 的权限。用户可能需要更新 default 服务账号的权限或指定另一个绑定了正确角色的服务账号。
kubectl create clusterrolebinding flink-role-binding-default --clusterrole=edit --serviceaccount=default:default
如果不想使用 default 服务帐户，可以使用以下命令创建新的 flink-service-account 服务帐户并设置角色绑定。然后使用 config 选项 -Dkubernetes.service-account=flink-service-account 使 JobManager pod 使用 flink-service-account 服务帐户来创建/删除 TaskManager pod。
kubectl create serviceaccount flink-service-account
kubectl create clusterrolebinding flink-role-binding-flink --clusterrole=edit --serviceaccount=default:flink-service-account
构建镜像
这里使用 Application Mode 模式在生产环境可以为应用提供更好的隔离。flink on k8s 要求代码与 flink 镜像绑定在一起，Application Mode 确保在应用程序终止后正确清理所有 Flink 组件。
编写代码
public class K8sDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<String> source = env.addSource(new SourceFunction<String>() {private volatile boolean isRunning = true;@Overridepublic void run(SourceContext<String> ctx) throws Exception {while (isRunning) {ctx.collect(UUID.randomUUID().toString());TimeUnit.SECONDS.sleep(1);}}@Overridepublic void cancel() {isRunning = false;}});source.print();env.execute();}
}
编写dockerfile
使用 flink 社区提供的基础 docker 镜像
FROM flink:1.13.6-scala_2.11-java8
RUN mkdir -p $FLINK_HOME/jobs
COPY k8s-demo-1.0-SNAPSHOT.jar $FLINK_HOME/jobs/flink-on-k8s.jar
构建镜像
docker build -t flink-on-k8s-demo .
提交任务
export一下k8s的KUBECONFIG
export KUBECONFIG=/etc/rancher/k3s/k3s.yaml
提交任务
bin/flink run-application \
--target kubernetes-application \
-Dkubernetes.cluster-id=my-first-application-cluster \
-Dkubernetes.service-account=flink-service-account \
-Dkubernetes.container.image=registry.cn-hangzhou.aliyuncs.com/czs-projects/flink-on-k8s-demo:v1.0 \
local:///opt/flink/jobs/k8s-demo-1.0-SNAPSHOT.jar
其中，根据RABC权限，启动失败要加上指定参数，否则可不加
-Dkubernetes.cluster-id=my-first-flink-cluster
-Dkubernetes.service-account=flink-service-account
报错To use support for EC Keys you must explicitly add this dependency to classpath将jar包加入flink client的目录下
之后执行成功日志
