网站开发技术合同,东台做网站的公司,杭州设计网站最好的公司,php企业网站开发教程星光下的赶路人star的个人主页 你必须赢过#xff0c;才可以说不在乎输赢 文章目录 1、Flink部署1.1 集群角色1.2 Flink集群搭建1.2.1 集群启动1.2.2 向集群提交作业 1.3 部署模式1.3.1 会话模式#xff08;Session Mode#xff09;1.3.2 单作业模式#xff08;Per-Job Mod… 星光下的赶路人star的个人主页 你必须赢过才可以说不在乎输赢 文章目录 1、Flink部署1.1 集群角色1.2 Flink集群搭建1.2.1 集群启动1.2.2 向集群提交作业 1.3 部署模式1.3.1 会话模式Session Mode1.3.2 单作业模式Per-Job Mode1.3.3 应用模式Application Mode 1.4 Standalone运行模式1.4.1 会话模式部署1.4.2 单作业模式部署1.4.3 应用模式部署 1.5 Yarn运行模式非常重要1.5.1 相关准备和配置1.5.2 会话模式部署1.5.3 单作业模式部署1.5.4 应用模式部署 1.6 K8S运行模式了解1.7 历史服务器 1、Flink部署
1.1 集群角色
Flink提交作业和执行任务需要几个关键组件
客户端Client代码由客户端获取并做转换之后提交给JobManagerJobManager就是Flink集群里的“管事人”对作业进行中央调度管理而它获取到要执行的作业后会进一步处理转换然后分发给众多的TaskManager。TaskManager就是真正的“干活的人”,数据的处理操作都是它们来做的。 注意Flink是一个非常灵活的处理框架它支持多种不同的部署场景还可以和不同的资源管理平台方便地集成。
1.2 Flink集群搭建
1.2.1 集群启动
0、集群规划
节点服务器hadoop102hadoop103hadoop104角色JobManager、TaskManagerTaskManagerTaskManager
具体安装部署步骤如下 1、下载并解压安装包 1下载直接去官网下安装包flink-1.17.0-bin-scala_2.12.tgz将该jar包上传到hadoop102节点服务器的/opt/software路径上。 2在/opt/software路径上解压安装包到/opt/module路径上
tar -zxvf flink-1.17.0-bin-scala_2.12.tgz -C /opt/module/2、修改集群配置 1进入conf路径修改flink-conf.yaml文件指定hadoop102节点服务器为JobManager
vim flink-conf.yaml修改如下内容:
# JobManager节点地址.
jobmanager.rpc.address: hadoop102
jobmanager.bind-host: 0.0.0.0
rest.address: hadoop102
rest.bind-address: 0.0.0.0
# TaskManager节点地址.需要配置为当前机器名
taskmanager.bind-host: 0.0.0.0
taskmanager.host: hadoop1022修改workers文件指定hadoop102、hadoop103和hadoop104为TaskManager vim workers修改如下内容
hadoop102
hadoop103
hadoop1043修改master文件
vim masters修改内容如下:
hadoop102:80814另外在flink-conf.yaml文件中还可以对集群中的JobManager和TaskManager组件进行优化配置主要配置项如下
jobmanager.memory.process.size对JobManager进程可使用到的全部内存进行配置包括JVM元空间和其他开销默认为1600M可以根据集群规模进行适当调整。taskmanager.memory.process.size对TaskManager进程可使用到的全部内存进行配置包括JVM元空间和其他开销默认为1728M可以根据集群规模进行适当调整。taskmanager.numberOfTaskSlots对每个TaskManager能够分配的Slot数量进行配置默认为1可根据TaskManager所在的机器能够提供给Flink的CPU数量决定。所谓Slot就是TaskManager中具体运行一个任务所分配的计算资源。parallelism.defaultFlink任务执行的并行度默认为1。优先级低于代码中进行的并行度配置和任务提交时使用参数指定的并行度数量。 3、分发安装目录 1配置修改完毕后将Flink安装目录发给另外两个节点服务器。
xsync flink-1.17.0/2修改hadoop103的 taskmanager.host 修改如下内容
# TaskManager节点地址.需要配置为当前机器名
taskmanager.host: hadoop1033修改hadoop104的 taskmanager.host和上面差不多只是改为hadoop104而已 4、启动集群 1在hadoop102节点服务器上执行start-cluster.sh启动Flink集群 bin/start-cluster.sh2查看进程情况 5、访问WebUI 启动成功后同样可以访问http://hadoop102:8081对Flink集群和任务进行监控管理。 这样可以明显看到当前集群的TaskManager数量为3由于默认每个TaskManager的Slot数量为1所以总Slot数和可用Slot数都为3。
1.2.2 向集群提交作业
在上一章中我们已经编写读取socket发送的单词并统计单词的个数程序案例。本节我们将以该程序为例演示如何将任务提交到集群中进行执行。具体步骤如下。
1、环境准备 在hadoop102中执行以下命令启动netcat。
nc -lk hadoop102 77772、程序打包 1在我们编写的Flink入门程序的pom.xml文件中添加打包插件的配置具体如下
buildpluginsplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-shade-plugin/artifactIdversion3.2.4/versionexecutionsexecutionphasepackage/phasegoalsgoalshade/goal/goalsconfigurationartifactSetexcludesexcludecom.google.code.findbugs:jsr305/excludeexcludeorg.slf4j:*/excludeexcludelog4j:*/exclude/excludes/artifactSetfiltersfilter!-- Do not copy the signatures in the META-INF folder.Otherwise, this might cause SecurityExceptions when using the JAR. --artifact*:*/artifactexcludesexcludeMETA-INF/*.SF/excludeexcludeMETA-INF/*.DSA/excludeexcludeMETA-INF/*.RSA/exclude/excludes/filter/filterstransformers combine.childrenappendtransformerimplementationorg.apache.maven.plugins.shade.resource.ServicesResourceTransformer/transformer/transformers/configuration/execution/executions/plugin/plugins
/build2插件配置完毕后可以使用IDEA的Maven工具执行package命令出现如下提示即表示打包成功。 打包完成后在target目录下即可找到所需JAR包JAR包会有两个FlinkDemo-1.0-SNAPSHOT.jar和original-FlinkDemo-1.0-SNAPSHOT.jar因为集群中已经具备任务运行所需的所有依赖所以建议使用FlinkDemo-1.0-SNAPSHOT.jar。
3、在WebUI上提交作业 1任务打包完成后我们打开Flink的WEB UI页面在右侧导航栏点击“Submit New Job”然后点击按钮“ Add New”选择要上传运行的JAR包如下图所示。 jar包上传完成如下图所示 2点击该jar包出现任务配置页面进行相应的配置 主要配置程序入口主类的全类名任务运行的并行度任务运行所需的配置参数和保存点路径等如下图所示配置完成后即可点击按钮“Submit”将任务提交到集群运行。
3任务提交成功之后可点击左侧导航栏的“Running Jobs”查看程序运行列表情况。
4测试 ①在socket端口中输入hello
②按图顺序 ③点击Stdout就可以看到hello单词的统计 4点击该任务可以查看任务运行的具体情况也可以通过点击“Cancel Job”结束任务运行。 4、命令行提交作业 除了通过WebUI界面提交任务之外也可以直接通过命令来提交任务。这里为方便起见我们可以先把jar包直接上传到目录flink-1.17.0下。 1首先要启动集群
bin/start-cluster.sh2在hadoop102中执行以下命令启动netcat。
nc -lk 99993将flink程序运行jar包上传到/opt/module/flink-1.17.0路径 4进入到flink的安装路径下在命令行使用flink run命令提交作业。
bin/flink run -m hadoop102:8081 -c com.zhm.wordcount.Demo03_UnBoundedStreamProcess ./FlinkDemo-1.0-SNAPSHOT.jar 这里的参数 -m指定了提交到的JobManager-c指定了入口类。 5在浏览器中打开Web UIhttp://hadoop102:8081查看应用执行情况。用netcat输入数据可以在TaskManager的标准输出Stdout看到对应的统计结果。 6数据模拟和实现和WebUI提交是一样的
1.3 部署模式
在一些应用场景中对于集群资源分配和占用的方式可能会有特定的需求。Flink为各种场景提供了不同的部署模式主要有以下三种模式会话模式Session Mode、单作业模式Per-Job Mode、应用模式Application Mode。 它们的区别主要在于集群的生命周期以及资源的分配方式以及应用的main方法到底在哪里执行—客户端Client还是JobManager。
1.3.1 会话模式Session Mode
会话模式其实最符合常规思维。我们需要先启动一个集群保持一个会话在这个会话中通过客户端提交作业。集群启动时所有资源就都已经确定所以所有提交的作业会竞争集群中的资源。 会话模式比较适合于单个规模小、执行时间短的大量作业。
1.3.2 单作业模式Per-Job Mode
会话模式因为资源共享会导致很多问题所以为了更好的隔离资源我们可以考虑为每个提交的作业启动一个集群这就是所谓的单作业Per-Job模式。 作业完成后集群就会关闭所有资源也会释放。 这些特性使得单作业模式在生产环境运行更加稳定所以是实际应用的首选模式。 需要注意的是Flink本身无法直接这样运行所有单作业模式一般需要借助一些资源管理框架爱来启动集群比如Yarn、KubernetsK8S。
1.3.3 应用模式Application Mode
前面两个模式下应用代码都是在客户端上执行的然后由于客户端提交给JobManager的。但是这种方式客户端需要占用大量网络带宽去下载依赖和二进制数据发送给JobManager加上很多情况下我们提交作业用的是同一个客户端就会加重客户端所在节点的资源消耗。 所以解决办法就是我们不要客户端了直接把应用提交到JobManager上运行。而这也就代表着我们需要为每一个提交的应用单独启动一个JobManager也就是创建一个集群。这个JobManager只为执行这一个应用而存在执行结束之后JobManager也就关闭了这就是所谓的应用模式。 应用模式和单作业模式都是提交作业之后才创建集群的单作业模式是通过客户端来提交的客户端解析出的每一个作业对应一个集群而应用模式下是直接由JobManager执行应用程序的。
这里说的部署模式是比较抽象的概念。实际应用的时候一般需要和资源管理平台结合起来选择特定的模式来分配资源、部署应用。接下来我们就针对不同的资源提供者的场景。
1.4 Standalone运行模式
独立模式是独立运行的不依赖任何外部的资源管理平台当然独立也是有代价的如果资源不足或者出现故障没有自动扩展或重分配资源的保证必须手动处理。所以独立模式一般只用在开发测试或作业非常少的场景下。
1.4.1 会话模式部署
在1.3.2 节就是用的Standalone集群的会话模式部署。 提前启动集群并通过Web页面客户端提交任务可以多个任务但是集群资源固定
1.4.2 单作业模式部署
Flink的Standalone集群并不支持单作业模式部署。因为单作业模式需要借助一些资源管理平台。
1.4.3 应用模式部署
应用模式下不会提前创建集群所以不能调用start-cluster.sh脚本。我们可以使用同样在bin目录下的standalone-job.sh来创建一个JobManager。 具体步骤如下 1环境准备在hadoop102中执行以下命令启动netcat。
nc -lk hadoop102 99992进入到Flink的安装路径下将应用程序的jar包放到lib/目录下。 3执行以下命令启动JobManager。
bin/standalone-job.sh start --job-classname com.zhm.wordcount.Demo03_UnBoundedStreamProcess 这里我们直接指定作业入口类脚本会到lib目录下扫描所有的jar包。 4同样是使用bin目录下的脚本启动TaskManager。
bin/taskmanager.sh start5在hadoop102上模拟发送单词数据。 6在hadoop102:8081地址中观察输出数据 7如果希望停掉集群同样可以使用脚本命令如下 bin/taskmanager.sh stopbin/standalone-job.sh stop1.5 Yarn运行模式非常重要
Yarn上部署的过程是客户端把Flink应用提交给Yarn的ResourceManagerYarn的ResourceManager会向Yarn的NodeManager申请容器Container。这些容器上Flink会部署JobManager和TaskManager的实例从而启动集群。Flink会根据运行在JobManager·上的作业所需要的槽数动态分配TaskManager资源。
1.5.1 相关准备和配置
在Flink任务不是至Yarn集群之前需要确认集群是否按照有Hadoop保证Hadoop版本至少在2.2以上并且集群中安装有HDFS服务。 1配置环境变量增加环境变量配置如下
sudo vim /etc/profile.d/my_env.sh#添加内容
HADOOP_HOME/opt/module/hadoop-3.3.4
export PATH$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
export HADOOP_CONF_DIR${HADOOP_HOME}/etc/hadoop
export HADOOP_CLASSPATHhadoop classpath2启动Hadoop集群包括HDFS和Yarn
start-dfs.sh
start-yarn.sh3在hadoop102执行以下命令启动netcat
nc -lk hadoop102 99991.5.2 会话模式部署
Yarn的会话模式与独立集群不一样需要首先申请一个Yarn会话Yarn Session来启动Flink集群具体步骤如下: 1、启动集群 1启动Hadoop集群HDFS、YARN。 2执行脚本命令向YARN集群申请资源开启一个YARN会话启动Flink集群。
bin/yarn-session.sh -nm test可用参数解读
-d分离模式如果你不想让Flink YARN客户端一直前台运行可以使用这个参数即使关掉当前对话窗口YARN session也可以后台运行。-jm–jobManagerMemory配置JobManager所需内存默认单位MB。-nm–name配置在YARN UI界面上显示的任务名。-qu–queue指定YARN队列名。-tm–taskManager配置每个TaskManager所使用内存。
注意Flink1.11.0版本不再使用-n参数和-s参数分别指定TaskManager数量和slot数量Yarn会按照需求动态分配TaskManager和Slot数量。所以从这个意义上来说Yarn的会话模式也不会把集群资源固定同样是动态分布的。
Yarn Session启动之后会给出一个WebUI地址以及一个Yarn ApplicationId如下所示用户可以通过Web UI或者命令行两种方式提交作业。
2、提交作业 1通过Web UI提交作业 这种方式比较简单与上文述Standalone部署模式基本一样。 2通过命令行提交作业 ① 将FlinkTutorial-1.0-SNAPSHOT.jar任务上传至集群。 ② 执行以下命令将该任务提交到已经开启的Yarn-Session中运行。
bin/flink run -c com.zhm.wordcount.Demo03_UnBoundedStreamProcess FlinkDemo-1.0-SNAPSHOT.jar 客户端可以自行确定JobManager的地址也可以通过-m或者-jobmanager参数指定JobManager的地址JobManager的地址在Yarn Session的启动页面可以找到 ③ 任务提交成功后可在YARN的Web UI界面查看运行情况。hadoop103:8088。 从上图中可以看到我们创建的Yarn-Session实际上是一个Yarn的Application并且有唯一的Application ID。 ④也可以通过Flink的Web UI页面查看提交任务的运行情况如下图所示。
1.5.3 单作业模式部署
在Yarn环境中由于有了外部平台做资源调度所以我们也可以直接向Yarn提交一个单独的作业从而启动一个Flink集群。 1执行命令提交作业
bin/flink run -d -t yarn-per-job -c com.zhm.wordcount.Demo03_UnBoundedStreamProcess FlinkDemo-1.0-SNAPSHOT.jar 2在YARN的ResourceManager界面查看执行情况。 点击(ApplicationMaster)可以打开Flink Web UI页面进行监控如下图所示 3可以使用命令行查看或取消作业命令如下。
##查看当前job的idbin/flink list -t yarn-per-job -Dyarn.application.idapplication_XXXX_YY## 取消jobbin/flink cancel -t yarn-per-job -Dyarn.application.idapplication_XXXX_YY jobId
这里的application_XXXX_YY是当前应用的ID是作业的ID。注意如果取消作业整个Flink集群也会停掉。
1.5.4 应用模式部署
应用模式同样非常简单与单作业模式类似直接执行flink run-application命令即可。 1、命令行提交 1执行命令提交作业。
bin/flink run-application -t yarn-application -c com.zhm.wordcount.Demo03_UnBoundedStreamProcess FlinkDemo-1.0-SNAPSHOT.jar 2在命令行中查看或取消作业。和单作业部署是一样的 2、上传HDFS提交 可以通过yarn.provided.lib.dirs配置选项指定位置将flink的依赖上传到远程。 1上传flink的lib和plugins到HDFS上
## 创建HDFS目录
fs -mkdir /flink-dist
fs -put lib/ /flink-dist##上传文件到HDFS目录
hadoop fs -put plugins/ /flink-dist2上传自己的jar包到HDFS
hadoop fs -mkdir /flink-jars
hadoop fs -put FlinkDemo-1.0-SNAPSHOT.jar /flink-jars
3提交作业 bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirshdfs://hadoop102:8020/flink-dist -c com.zhm.wordcount.Demo03_UnBoundedStreamProcess hdfs://hadoop102:8020/flink-jars/FlinkDemo-1.0-SNAPSHOT.jar 这种方式上Flink本身的依赖和用户jar可以预先上传到HDFS而不需要单独发送到集群这就使得作业提交更加轻量了。
1.6 K8S运行模式了解
容器化部署是如今业界流行的一项技术基于Docker镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的KubernetesK8S而Flink也在最近的版本中支持了K8S部署模式。基本原理与Yarn是类似的。
1.7 历史服务器
运行 Flink job 的集群一旦停止只能去 yarn 或本地磁盘上查看日志不再可以查看作业挂掉之前的运行的 Web UI很难清楚知道作业在挂的那一刻到底发生了什么。如果我们还没有 Metrics 监控的话那么完全就只能通过日志去分析和定位问题了所以如果能还原之前的 Web UI我们可以通过 UI 发现和定位一些问题。 Flink提供了历史服务器用来在相应的 Flink 集群关闭后查询已完成作业的统计信息。我们都知道只有当作业处于运行中的状态才能够查看到相关的WebUI统计信息。通过 History Server 我们才能查询这些已完成作业的统计信息无论是正常退出还是异常退出。 此外它对外提供了 REST API它接受 HTTP 请求并使用 JSON 数据进行响应。Flink 任务停止后JobManager 会将已经完成任务的统计信息进行存档History Server 进程则在任务停止后可以对任务统计信息进行查询。比如最后一次的 Checkpoint、任务运行时的相关配置。
1、创建储存目录
hadoop fs -mkdir -p /logs/flink-job2、在flink-config.yaml中添加如下配置
jobmanager.archive.fs.dir: hdfs://hadoop102:8020/logs/flink-job
historyserver.web.address: hadoop102
historyserver.web.port: 8082
historyserver.archive.fs.dir: hdfs://hadoop102:8020/logs/flink-job
historyserver.archive.fs.refresh-interval: 50003、启动历史服务器
bin/historyserver.sh start4、停止历史服务器
bin/historyserver.sh stop5、在浏览器地址栏输入http://hadoop102:8082 查看已经停止的 job 的统计信息 您的支持是我创作的无限动力 希望我能为您的未来尽绵薄之力 如有错误谢谢指正若有收获谢谢赞美
