Flink on YARN时,如何确定TaskManager数

摘要:
JobManager和TaskManager本质上都是JVM进程。为了提高Flink程序的运行效率和资源利用率,Flink在TaskManager中实现了任务槽。所以,可以得出FlinkonYARN时,TaskManager的数量就是:max/yarnslots。例如,一个最大并行度为10,每个TaskManager有两个任务槽的作业,就会启动5个TaskManager,如WebUI所示。

转自:https://www.jianshu.com/p/5b670d524fa5

答案写在最前面:Job的最大并行度除以每个TaskManager分配的任务槽数。

问题

Flink 1.5 Release Notes中,有这样一段话,直接上截图。

Flink on YARN时,如何确定TaskManager数第1张

这说明从1.5版本开始,Flink on YARN时的容器数量——亦即TaskManager数量——将由程序的并行度自动推算,也就是说flink run脚本的-yn/--yarncontainer参数不起作用了。那么自动推算的规则是什么呢?要弄清楚它,先来复习Flink的并行度(Parallelism)和任务槽(Task Slot)。

并行度(Parallelism)

与Spark类似地,一个Flink Job在生成执行计划时也划分成多个Task。Task可以是Source、Sink、算子或算子链(算子链有点意思,之后会另写文章详细说的)。Task可以由多线程并发执行,每个线程处理Task输入数据的一个子集。而并发的数量就称为Parallelism,即并行度。

Flink程序中设定并行度有4种级别,从低到高分别为:算子级别、执行环境(ExecutionEnvironment)级别、客户端(命令行)级别、配置文件(flink-conf.yaml)级别。实际执行时,优先级则是反过来的,算子级别最高。简单示例如下。

  • 算子级别
dataStream.flatMap(new SomeFlatMapFunction()).setParallelism(4);
  • 执行环境级别
streamExecutionEnvironment.setParallelism(4);
  • 命令行级别
bin/flink -run --parallelism 4 example-0.1.jar
  • flink-conf.yaml级别
parallelism.default: 4

任务槽(Task Slot)

Flink运行时由两个组件组成:JobManager与TaskManager,与Spark Standalone模式下的Master与Worker是同等概念。从官网抄来的图如下所示,很容易理解。

Flink on YARN时,如何确定TaskManager数第2张

JobManager和TaskManager本质上都是JVM进程。为了提高Flink程序的运行效率和资源利用率,Flink在TaskManager中实现了任务槽(Task Slot)。任务槽是Flink计算资源的基本单位,每个任务槽可以在同一时间执行一个Task,而TaskManager可以拥有一个或者多个任务槽。

任务槽可以实现TaskManager中不同Task的资源隔离,不过是逻辑隔离,并且只隔离内存,亦即在调度层面认为每个任务槽“应该”得到taskmanager.heap.size的N分之一大小的内存。CPU资源不算在内。

TaskManager的任务槽个数在使用flink run脚本提交on YARN作业时用-ys/--yarnslots参数来指定,另外在flink-conf.yaml文件中也有默认值taskManager.numberOfTaskSlots。一般来讲,我们设定该参数时可以将它理解成一个TaskManager可以利用的CPU核心数,因此也要根据实际情况(集群的CPU资源和作业的计算量)来确定。

确定TaskManager数

以Flink自带示例中简化的WordCount程序为例:

    // 执行环境并行度设为6
    env.setParallelism(6);
    // Source并行度为1
    DataStream<String> text = env
      .readTextFile(params.get("input"))
      .setParallelism(1);
    DataStream<Tuple2<String, Integer>> counts = text
      .flatMap(new Tokenizer())
      .keyBy(0)
      .sum(1);
    counts.print();

--yarnslots 3参数来执行,即每个TaskManager分配3个任务槽。TaskManager、任务槽和任务的分布将如下图所示,方括号内的数字为并行线程的编号。

Flink on YARN时,如何确定TaskManager数第3张
图来自http://wuchong.me/blog/2016/05/09/flink-internals-understanding-execution-resources/,致敬

由图中可以看出,由于算子链机制的存在,KeyAgg与Sink操作链接在了一起,作为一个Task来执行。

Flink允许任务槽共享,即来自同一个Job的不同Task的Sub-Task(理解为Task的子集就行)进入同一个槽位,因此在图中也可以见到任务槽X中同时存在FlatMap[X]与KeyAgg[X]+Sink[X]。任务槽共享有两点好处:

  • 能够让每个Task的Sub-Task都均摊到不同的TaskManager,避免负载倾斜。
  • 不需要再计算App一共需要起多少个Task,因为作业需要的任务槽数量肯定等于Job中最大的并行度。

所以,可以得出Flink on YARN时,TaskManager的数量就是:max(parallelism) / yarnslots(向上取整)。例如,一个最大并行度为10,每个TaskManager有两个任务槽的作业,就会启动5个TaskManager,如Web UI所示。

Flink on YARN时,如何确定TaskManager数第4张

免责声明:文章转载自《Flink on YARN时,如何确定TaskManager数》仅用于学习参考。如对内容有疑问,请及时联系本站处理。

上篇Time Series in pandasCKEditor粘贴图片自动上传到服务器(JS版)下篇

宿迁高防,2C2G15M,22元/月;香港BGP,2C5G5M,25元/月 雨云优惠码:MjYwNzM=

相关文章

有赞 Flink 实时任务资源优化探索与实践

简介: 目前有赞实时计算平台对于 Flink 任务资源优化探索已经走出第一步。 随着 Flink K8s 化以及实时集群迁移完成,有赞越来越多的 Flink 实时任务运行在 K8s 集群上,Flink K8s 化提升了实时集群在大促时弹性扩缩容能力,更好的降低大促期间机器扩缩容的成本。同时,由于 K8s 在公司内部有专门的团队进行维护, Flink K8s...

Spark优化之二:集群上运行jar程序,状态一直Accepted且不停止不报错

如果运行Spark集群时状态一直为Accepted且不停止不报错,比如像下面这样的情况: 15/06/14 11:33:33 INFO yarn.Client: Application report for application_1434263747091_0023 (state: ACCEPTED) 15/06/14 11:33:34 INFO yar...

idea搭建flink环境

1.打开idea,选择new project,跳转至如下界面: 2.单击next 下一步,输入项目名称,单击finish完成 3.在项目main项目下新建scala目录,并新建文件,文件名称为 hello.scala   4.新建文件之后,进入helllo.scala文件,会弹出添加sdk,添加自己版本的scala就可以。完成之后可以新建添加如下代码,...

Flink1.13.2版本 Standalone 模式部署

Flink的部署有3种模式,分别是local模式、Standalone模式、yarn模式。其中local就是单机模式,一般来说用于本地开发测试;Standalone跟yarn模式都可以支撑集群部署、实现HA,但是两者在任务分配机制、内存管理等内容上有比较大的差异。一般在处理计算数据量级非常大的生产环境,使用flink on yarn的模式更多一些。 我们的...

更改hadoop集群yarn的webui中的开始时间和结束时间为本地时间

  yarn集群的webui地址为:http://rm:8088   执行任务后,任务的开始时间和结束时间都是utc时间,查看很不方便。   查找相关资料发现hadoop有补丁包,补丁地址:https://issues.apache.org/jira/browse/YARN-1998   补丁内容为: Index: hadoop-yarn-project/...

Flink--将表转换为DataStream或DataSet

A Table可以转换成a DataStream或DataSet。通过这种方式,可以在Table API或SQL查询的结果上运行自定义的DataStream或DataSet程序 将表转换为DataStream 有两种模式可以将 Table转换为DataStream: 1:Append Mode 将一个表附加到流上 2:Retract Mode 将表转换为流...