爱战网关键词挖掘查询工具,搜索引擎优化英文简称为,dede网站安装教程,怎么建立一个小说网站Flink On Yarn 模式
基于Yarn层面的架构类似 Spark on Yarn模式#xff0c;都是由Client提交App到RM上面去运行#xff0c;然后 RM分配第一个container去运行AM#xff0c;然后由AM去负责资源的监督和管理。需要说明的是#xff0c;Flink的Yarn模式更加类似Spark on Ya…Flink On Yarn 模式
基于Yarn层面的架构类似 Spark on Yarn模式都是由Client提交App到RM上面去运行然后 RM分配第一个container去运行AM然后由AM去负责资源的监督和管理。需要说明的是Flink的Yarn模式更加类似Spark on Yarn的cluster模式在cluster模式中dirver将作为AM中的一个线程去运行。Flink on Yarn模式也是会将JobManager启动在container里面去做个driver类似的任务调度和分配Yarn AM与Flink JobManager在同一个Container中这样AM可以知道Flink JobManager的地址从而AM可以申请Container去启动Flink TaskManager。待Flink成功运行在Yarn集群上Flink Yarn Client就可以提交Flink Job到Flink JobManager并进行后续的映射、调度和计算处理。
Fink on Yarn 的缺陷
【1】资源分配是静态的一个作业需要在启动时获取所需的资源并且在它的生命周期里一直持有这些资源。这导致了作业不能随负载变化而动态调整在负载下降时无法归还空闲的资源在负载上升时也无法动态扩展。 【2】On-Yarn模式下所有的container都是固定大小的导致无法根据作业需求来调整container的结构。譬如CPU密集的作业或需要更多的核但不需要太多内存固定结构的container会导致内存被浪费。 【3】与容器管理基础设施的交互比较笨拙需要两个步骤来启动Flink作业1.启动Flink守护进程2.提交作业。如果作业被容器化并且将作业部署作为容器部署的一部分那么将不再需要步骤2。 【4】On-Yarn模式下作业管理页面会在作业完成后消失不可访问。 【5】Flink推荐 per job clusters 的部署方式但是又支持可以在一个集群上运行多个作业的session模式令人疑惑。
在Flink版本1.5中引入了DispatcherDispatcher是在新设计里引入的一个新概念。Dispatcher会从Client端接受作业提交请求并代表它在集群管理器上启动作业。引入Dispatcher的原因主要有两点 【1】一些集群管理器需要一个中心化的作业生成和监控实例 【2】能够实现Standalone模式下JobManager的角色且等待作业提交。在一些案例中Dispatcher是可选的Yarn或者不兼容的kubernetes。
资源调度模型重构下的 Flink On Yarn 模式 客户端提交JobGraph以及依赖jar包到YarnResourceManager接着Yarn ResourceManager分配第一个container以此来启动AppMasterApplication Master中会启动一个FlinkResourceManager以及JobManagerJobManager会根据JobGraph生成的ExecutionGraph以及物理执行计划向FlinkResourceManager申请slotFlinkResoourceManager会管理这些slot以及请求如果没有可用slot就向Yarn的ResourceManager申请containercontainer启动以后会注册到FlinkResourceManager最后JobManager会将subTask deploy到对应container的 slot中去。
在有Dispatcher的模式下会增加一个过程就是Client会直接通过HTTP Server的方式然后用Dispatcher将这个任务提交到Yarn ResourceManager中。
新框架具有四大优势详情如下 【1】client直接在Yarn上启动作业而不需要先启动一个集群然后再提交作业到集群。因此client再提交作业后可以马上返回。 【2】所有的用户依赖库和配置文件都被直接放在应用的classpath而不是用动态的用户代码classloader去加载。 【3】container在需要时才请求不再使用时会被释放。 【4】“需要时申请”的container分配方式允许不同算子使用不同profile (CPU和内存结构)的container。
新的资源调度框架下 single cluster job on Yarn 流程介绍 single cluster job on Yarn模式涉及三个实例对象 【1】clifrontend Invoke App code生成StreamGraph然后转化为JobGraph 【2】YarnJobClusterEntrypointMaster 依次启动YarnResourceManager、MinDispatcher、JobManagerRunner三者都服从分布式协同一致的策略JobManagerRunner将JobGraph转化为ExecutionGraph然后转化为物理执行任务Execution然后进行deploydeploy过程会向 YarnResourceManager请求slot如果有直接deploy到对应的YarnTaskExecutiontor的slot里面没有则向Yarn的ResourceManager申请带container启动以后deploy。 【3】YarnTaskExecutorRunner (slave) 负责接收subTask并运行。
整个任务运行代码调用流程如下图 subTask在执行时是怎么运行的
调用StreamTask的invoke方法执行步骤如下 【1】initializeState()即operator的initializeState() 【2】openAllOperators()即operator的open()方法 【3】最后调用run方法来进行真正的任务处理
我们来看下flatMap对应的OneInputStreamTask的run方法具体是怎么处理的。
Override
protected void run() throws Exception {// 在堆栈上缓存处理器引用使代码更易于JITfinal StreamInputProcessorIN inputProcessor this.inputProcessor;while (running inputProcessor.processInput()) {// 所有的工作都发生在“processInput”方法中}
}最终是调用StreamInputProcessor的processInput()做数据的处理这里面包含用户的处理逻辑。
public boolean processInput() throws Exception {if (isFinished) {return false;}if (numRecordsIn null) {try {numRecordsIn ((OperatorMetricGroup) streamOperator.getMetricGroup()).getIOMetricGroup().getNumRecordsInCounter();} catch (Exception e) {LOG.warn(An exception occurred during the metrics setup., e);numRecordsIn new SimpleCounter();}}while (true) {if (currentRecordDeserializer ! null) {DeserializationResult result currentRecordDeserializer.getNextRecord(deserializationDelegate);if (result.isBufferConsumed()) {currentRecordDeserializer.getCurrentBuffer().recycleBuffer();currentRecordDeserializer null;}if (result.isFullRecord()) {StreamElement recordOrMark deserializationDelegate.getInstance();//处理watermarkif (recordOrMark.isWatermark()) {// handle watermark//watermark处理逻辑这里可能引起timer的triggerstatusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);continue;} else if (recordOrMark.isStreamStatus()) {// handle stream statusstatusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), currentChannel);continue;//处理latency watermark} else if (recordOrMark.isLatencyMarker()) {// handle latency markersynchronized (lock) {streamOperator.processLatencyMarker(recordOrMark.asLatencyMarker());}continue;} else {//用户的真正的代码逻辑// now we can do the actual processingStreamRecordIN record recordOrMark.asRecord();synchronized (lock) {numRecordsIn.inc();streamOperator.setKeyContextElement1(record);//处理数据streamOperator.processElement(record);}return true;}}}//这里会进行checkpoint barrier的判断和对齐以及不同partition 里面checkpoint barrier不一致时候的数据bufferfinal BufferOrEvent bufferOrEvent barrierHandler.getNextNonBlocked();if (bufferOrEvent ! null) {if (bufferOrEvent.isBuffer()) {currentChannel bufferOrEvent.getChannelIndex();currentRecordDeserializer recordDeserializers[currentChannel];currentRecordDeserializer.setNextBuffer(bufferOrEvent.getBuffer());}else {// Event receivedfinal AbstractEvent event bufferOrEvent.getEvent();if (event.getClass() ! EndOfPartitionEvent.class) {throw new IOException(Unexpected event: event);}}}else {isFinished true;if (!barrierHandler.isEmpty()) {throw new IllegalStateException(Trailing data in checkpoint barrier handler.);}return false;}}
}streamOperator.processElement(record)最终会调用用户的代码处理逻辑假如operator是StreamFlatMap的话。
Override
public void processElement(StreamRecordIN element) throws Exception {collector.setTimestamp(element);userFunction.flatMap(element.getValue(), collector);//用户代码
}
