本文分享自天翼云开发者社区《K8s Application模式下的flink任务执行精要》，作者：l****n
构键k8s集群
- 在这里，我们需要搭建一个K8S环境用于提供flink任务的运行时环境。在这里推荐使用kubeadm或者一些脚本工具搭建，可参考本自动k8s脚本工具。具体过程在这里省略，可以参考上述链接中的文档进行操作。
- 需要注意的是，我们需要在相应用户的目录下提供一个kubeconfig文件，如下图所示，通过该文件，StreamPark才能顺利地调用K8S客户端提交任务，该config的内容为与K8S的ApiServer进行连接时需要使用的信息。
提供flink运行任务的环境
- 将kubeconfig提供出来，供flink客户端调用
- 在这里主要提供一个供flink使用命名空间、和sa
- # 创建namespace<br>kubectl create ns flink-dev<br># 创建serviceaccount<br>kubectl create serviceaccount flink-service-account -n flink-dev<br># 用户授权<br>kubectl create clusterrolebinding flink-role-binding-flink --clusterrole=edit --serviceaccount=flink-dev:flink-service-account
下载flink客户端
flink客户端是控制flink的核心，需要下载并部署- wget https://archive.apache.org/dist/flink/flink-1.14.3/flink-1.14.3-bin-scala_2.12.tgz<br>tar -xf flink-1.14.3-bin-scala_2.12.tgz
任务jar生成过程
在这里，主要提供一个flink任务案例供flink k8s application进行调用
- 开发java代码，供使用，本示例项目较为简单，仅为将数据输出至mysql中，调用mysql-connector进行实现
- package cn.ctyun.demo;<br><br>import org.apache.flink.api.java.utils.ParameterTool;<br>import org.apache.flink.connector.jdbc.JdbcConnectionOptions;<br>import org.apache.flink.connector.jdbc.JdbcSink;<br>import org.apache.flink.streaming.api.datastream.DataStreamSource;<br>import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;<br><br>public class SinkToMySQL {<br> public static void main(String[] args) throws Exception {<br>// 从启动参数中获取连接信息<br> ParameterTool parameterTool = ParameterTool.fromArgs(new String[]{"url", "passwd", "user"});<br> StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();<br> env.setParallelism(1);<br><br><br> DataStreamSource<Event> stream = env.fromElements(<br> new Event("Mary", "./home", 1000L),<br> new Event("Bob", "./cart", 2000L),<br> new Event("Alice", "./prod?id=100", 3000L),<br> new Event("Alice", "./prod?id=200", 3500L),<br> new Event("Bob", "./prod?id=2", 2500L),<br> new Event("Alice", "./prod?id=300", 3600L),<br> new Event("Bob", "./home", 3000L),<br> new Event("Bob", "./prod?id=1", 2300L),<br> new Event("Bob", "./prod?id=3", 3300L));<br><br> stream.addSink(<br> JdbcSink.sink(<br> "INSERT INTO clicks (user, url) VALUES (?, ?)",<br> (statement, r) -> {<br> statement.setString(1, r.user);<br> statement.setString(2, r.url);<br> },<br> new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()<br> .withUrl(parameterTool.get("url"))<br> .withDriverName("com.mysql.jdbc.Driver")<br> .withUsername(parameterTool.get("user"))<br> .withPassword(parameterTool.get("passwd"))<br> .build()<br> )<br> );<br> env.execute();<br> }<br>}
- 项目打包
防止一些依赖缺失，这里使用fatjar的方式进行打包，注意，这里使用了jar-with-dependencies方法进行打包，即将依赖全部打入到相应的jar包中，这样可以防止平台上的flink因为以来缺失问题导致无法使用flink程序。maven相关的设置如下所示：- <plugin><br><groupId>org.apache.maven.plugins</groupId><br>maven-assembly-plugin</artifactId><br><version>3.0.0</version><br><configuration><br><descriptorRefs><br><descriptorRef>jar-with-dependencies</descriptorRef><br></descriptorRefs><br></configuration><br><executions><br><execution><br><id>make-assembly</id><br><phase>package</phase><br><goals><br><goal>single</goal><br></goals><br></execution><br></executions><br></plugin>
- 制作镜像，在这里通过官方镜像作为基础镜像进行构建，
使用docker进行镜像生成，使用命令为docker build -t /flink-demo-jar-job:1.0-SNAPSHOT .- FROM apache/flink:1.14.3-scala_2.12<br>RUN mkdir -p $FLINK_HOME/usrlib<br>COPY lib $FLINK_HOME/lib/<br>COPY flink-demo-jar-job-1.0-SNAPSHOT-jar-with-dependencies.jar $FLINK_HOME/usrlib/flink-demo-jar-job-1.0-SNAPSHOT-jar-with-dependencies.jar
- 推送镜像
这里推送镜像一般会推送到默认的dockerhub相应的仓库。如果需要push到自己的镜像仓，则需要修改相应的镜像前缀${docker_repository}为自己的镜像仓位置- docker push ${docker_repository}/flink-demo-jar-job:1.0-SNAPSHOT
k8s Application运行
Application模式架构
在k8s application模式下，用户只需要通过 Flink Client/CLI 启动作业。首先通过 K8s 启动 JobManager（deployment）的同时启动作业，然后通过 JobManager 内部的 K8sResourceManager 模块向 K8s 直接申请 TaskManager 的资源并启动，最后当 TM 注册到 JM 后作业就提交到 TM。用户在整个过程无需指定 TaskManager 资源的数量，而是由 JobManager 向 K8s 按需申请的。
启动命令
这里我们可以指定一定的运行参数，相关的参数设定方案请参考官方文档https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/deployment/config/#kubernetes- ./bin/flink run-application \<br>--target kubernetes-application \<br>-Dkubernetes.cluster-id=flink-cluster \<br># 指定容器启动的镜像(与之前提交的保持一致)<br>-Dkubernetes.container.image=****/flink-demo-jar-job:1.0-SNAPSHOT \<br>-Dkubernetes.jobmanager.replicas=1 \<br># 指定容器运行的命名空间<br>-Dkubernetes.namespace=flink-dev \<br>-Dkubernetes.jobmanager.service-account=flink-service-account \<br>-Dkubernetes.taskmanager.cpu=1 \<br>-Dtaskmanager.memory.process.size=4096mb \<br>-Dkubernetes.jobmanager.cpu=1 \<br>-Djobmanager.memory.process.size=4096mb \<br>-Dkubernetes.rest-service.exposed.type=NodePort \<br>-Dclassloader.resolve-order=parent-first \<br># yaml 模板,为解决hosts映射,后续可以通过编排此yaml文件,实现动态替换启动jar包、配置文件和持久化一些数据<br># -Dkubernetes.pod-template-file=/opt/flink-1.14.2/flink-templeta.yaml \<br># Main方法<br>-c cn.ctyun.demo.SinkToMySQL \<br># 启动Jar包和启动配置文件的绝对路径(容器内部,不是宿主机)<br>local:///usr/local/flink/lib/flink-realtime-1.0-SNAPSHOT.jar \<br># 如下将提供mysql的连接信息，通过参数的方式传递给jar包<br>--passwd ****** \<br>--user ******\<br>--url ******
PodTemplate主要是通过指定pod的启动样例，在podtemplate中可以指定域名、挂载路径、配置文件、初始化容器等信息，如下给出一个提供一个持久化保存日志的PodTemplate。- apiVersion: v1<br>kind: Pod<br>metadata:<br>name: jobmanager-pod-template<br>spec:<br>initContainers:<br> - name: artifacts-fetcher<br> image: artifacts-fetcher:latest<br> # Use wget or other tools to get user jars from remote storage<br> command: [ 'wget', 'https://path/of/StateMachineExample.jar', '-O', '/flink-artifact/myjob.jar' ]<br> volumeMounts:<br> - mountPath: /flink-artifact<br> name: flink-artifact<br>containers:<br> # Do not change the main container name<br> - name: flink-main-container<br> resources:<br> requests:<br> ephemeral-storage: 2048Mi<br> limits:<br> ephemeral-storage: 2048Mi<br> volumeMounts:<br> - mountPath: /opt/flink/volumes/hostpath<br> name: flink-volume-hostpath<br> - mountPath: /opt/flink/artifacts<br> name: flink-artifact<br> - mountPath: /opt/flink/log<br> name: flink-logs<br> # Use sidecar container to push logs to remote storage or do some other debugging things<br> - name: sidecar-log-collector<br> image: sidecar-log-collector:latest<br> command: [ 'command-to-upload', '/remote/path/of/flink-logs/' ]<br> volumeMounts:<br> - mountPath: /flink-logs<br> name: flink-logs<br>volumes:<br> - name: flink-volume-hostpath<br> hostPath:<br> path: /tmp<br> type: Directory<br> - name: flink-artifact<br> emptyDir: { }<br> - name: flink-logs<br> emptyDir: { }
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！ |
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
照妖镜
