文章目录一、离线任务二、相关理论1.Task Slots and Resources1.1. slots与资源的隔离和共享1.2 建议cpu和slot数关系2. tm的资源配置是否合适2.1. flink load problems2.2. 阿里 flink资源配置建议三、问题分析与解决1. 测试结果比对1.1. 任务11.2. 任务21.3. 任务3四、离线任务性能调优小结1. 任务稳定性2. 运行效率与1core多并发3. 内存调整Apache Flink作为分布式处理引擎用于对无界和有界数据流进行状态计算。其中实时任务用于处理无界流离线任务用于处理有界流。本文分析在使用flink处理离线任务时遇到的一些问题通过分析任务特点进行进行任务性能调优。通过本文你将掌握让大型离线任务运行稳定的能力同时能够通过分析离线任务运行特点降低任务运行资源消耗减少成本。下面我们进入正题一、离线任务对于平台处理的离线任务任务大都是处理从HDFS到HIVE的数据清洗任务。这类任务的特点是数据来一条处理一条所以任务大都是没有状态的。接下来我们看一个问题任务数据源: 在hdfs上有301个文件每个文件9.6G压缩后的大小总共大约240亿条数据。flink的处理逻辑对于每条数据通过正则去摄取目标数据。任务资源配置301并发、tm:10core、10slots、15G内存、jm: 10core、8G内存。那将会产生32个container运行在yarn中。问题描述任务运行的速度大概在1.2亿/min运行2小时50多分钟。但是任务会偶发的报hadoop集群的问题以至于后面这个任务少了几天数据任务都跑不下去。虽然看报错是hadoop集群的问题如下connection reset by peerEOFException: End of File Exception但其他类似的任务运行的很稳定“事出反常必有妖”本文尝试从内存、并发的角度分析任务的稳定性及任务运行速度等问题。二、相关理论1.Task Slots and Resources1.1. slots与资源的隔离和共享官网 Task Slots and ResourcesEach worker (TaskManager) is a JVM process, and may execute one or more subtasks in separate threads. To control how many tasks a worker accepts, a worker has so called task slots (at least one).Each task slot represents a fixed subset of resources of the TaskManager. A TaskManager with three slots, for example, will dedicate 1/3 of its managed memory to each slot. Slotting the resources means that a subtask will not compete with subtasks from other jobs for managed memory, but instead has a certain amount of reserved managed memory. Note that no CPU isolation happens here; currently slots only separate the managed memory of tasks.By adjusting the number of task slots, users can define how subtasks are isolated from each other. Having one slot per TaskManager means each task group runs in a separate JVM (which can be started in a separate container, for example). Having multiple slots means more subtasks share the same JVM. Tasks in the same JVM share TCP connections (via multiplexing) and heartbeat messages. They may also share data sets and data structures, thus reducing the per-task overhead.这里我们可以获取每一个tmtaskmanager都是一个jvm进程一个tm可能执行一个或多个subtask在各自的线程其实就是slot中。每一个slot代表tm固定的资源子集比如一个tm有3个slot那么将tm将均匀分配它的托管内存给每个slot。slot意味着subtask将不会占用其他subtasks的资源而是会拥有固定预留的托管内存。但注意这里没有cpu的隔离目前slots仅分离了托管内存。通过调整tm内slot的数量用户可以定义subtasks之间的隔离方式。比如当tm只有一个slot意味着每个task组都运行在单独JVM运行比如yarn的container。当tm有多个slot时意味着多个子任务共享同一个JVM。同一个JVM的subtasks共享TCP通过多路复用连接和心跳消息他们还可以共享数据集和数据结构从而减少每个任务的开销。所以推荐一个tm包含多个slot小结tm是一个jvm进程一个tm可能执行一个或多个subtask在各自的线程slot中。tm的托管内存资源会根据slot的数量而分开但是tm中的 TCP通过多路复用连接、心跳消息、共享数据集、数据结构和cpu资源各slot会共享。所以我们可以设置多个slot 在一个tm中来实现资源共享但一个tm中设置几个slot合适呢我们先带着疑问接着往下看1.2 建议cpu和slot数关系As a rule-of-thumb, a good default number of task slots would be the number of CPU cores. With hyper-threading, each slot then takes 2 or more hardware thread contexts.根据经验一个好的任务插槽的默认数量应该是CPU内核的数量。但如果考虑到使用超线程每个插槽将接受2个或更多硬件线程上下文。有超线程的机器可以建议设置numOfslots 2 * numOfcores 没有超线程的机器建议设置numOfslot numOfcore。这里可以思考没有超线程就不能设置numOfslots numOfcores了吗毕竟我们的任务是IO密集型的任务这个可以测试一下。参考Ideal Number of Task Slots2. tm的资源配置是否合适目前配置的tm是:10core、10slot、15G但是跑上述任务时任务不稳定这里我查到一些网络上的信息2.1. flink load problemsflink load problemsThere’s also the consideration of Flink’s scheduling algorithm. We’ve frequently run into problems where, with multiple hosts running one large task manager a piece, all jobs get scheduled to one host, which can cause load problems.Flink的调度算法我们经常遇到这样的问题:当多个主机tm同时运行一个大型任务管理器时所有作业都被调度到一个主机上这可能会导致负载问题。We ended up makingmultiple smaller task managers per hostand jobs seem to be distributed better (although they still cluster on one node often).我们最终在每个主机上创建了多个较小的任务管理器并且作业似乎可以更好地分布(尽管它们仍然经常聚集在一个节点上)。简单的总结上面的经验就是调小tm的资源cpu和memory作业可以更好的分布。场景对应的应该是当任务的数据集中在个别节点的时候。2.2. 阿里 flink资源配置建议阿里 flink资源配置建议TaskManager资源设置不宜过小也不宜过大:如果单个TaskManager资源过小则可能影响其上作业的稳定性并且由于其Slot数目不多无法有效平摊TaskManager的开销降低了资源的利用效率。如果单个TaskManager资源过大则TaskManager上运行的作业数会很多一旦TaskManager发生单点故障影响面会很大。小结当tm设置的资源过大时遇到单点问题影响面很大。目前看在部署taskmanager 10core,15G 时tm资源设置的大了造成的单点故障的概率提高。三、问题分析与解决总体的调整思路目前10core,15G 的设置导致当任务规模301个文件 241亿数据到达一定水平时任务运行的将变得不稳定所以这里调小Tm的cores、memory。因为任务的特点是IO密集型所以可以考虑1个cpu对应多个slot个数这里flink建议是2倍但需要测试。当减小每个Tm的资源时Tm的个数将会增加在相同任务下这时需要考虑Jobmanager的调度压力和管理压力是否对任务运行的稳定性和效率有所影响。1. 测试结果比对1.1. 任务1数据源: 在hdfs上有301个文件每个文件9.6G压缩后的大小接收后11TB总共大约240亿条数据、处理对于每条数据通过正则去摄取目标数据。任务初始配置301并发、tm:10core、10slots、15G内存、jm: 10core、8G内存。那将会产生32个container运行在yarn中。序号任务设置描述运行结果1301并发、tm:10core 10slots 15G1.18亿/min、71亿/h、偶发性失败2301并发、tm:1core 2slots 4G0.5亿/min242亿运行了6.5小时3301并发、tm:1core 2slots 8G2/3亿/min、44亿/h4301并发、tm:1core 4slots 8G0.3亿/min、18亿/h速度更慢了5301并发、tm:3core 6slots 12G2/3亿/min、44亿/h6301并发、tm:5core 10slots 15G5/6亿/min、50亿/h7301并发、tm:6core 6slots 12G/9G1.18亿/min、71亿/h、2个多小时跑完8301并发、tm:4core 4slots 6G1.18亿/min、71亿/h、2个多小时跑完9301并发、tm:4core 4slots 4G1.18亿/min、71亿/h、2个多小时跑完;测试多次稳定跑完现在从3个方面讨论任务运行的情况1.速度从第235运行结果对比可以看出yarn集群对于1core支持多并发的速度没有达到超线程效果、或对于hdfs到hive的io密集型任务没有收获很好的效果从第4运行结果看出当1core 对应 4并发时速度下降接近一半2.内存使用与资源共享从5、6运行结果对比看出随着一个tm的slot数的增多速度有所提升提升不高这里可以暂时认为是tm内的 TCP连接、心跳消息、共享数据集、数据结构和cpu资源等起到了共享的作用。从7、8、9运行结果对比可以看出运行速度基本都到了最高峰此时tm的共享、内存的提高均没有提升flink的运行速度。3.稳定性从9运行结果可以看出运行速度达到了最快且多次运行后都能稳定运行完。之前遇到的任务不稳定相关问题见【报错】flink源码分析 has no more allocated slots与思考1.2. 任务2数据源: 在hdfs上有1000多个文件每个文件1G压缩后的大小(每个文件有423467条)总共大约4.5亿条数据处理对于每条数据通过正则去摄取目标数据。序号任务设置描述运行结果11000并发、tm:4core 4slots 6G共有251个container部署花费了20分钟之多。。。总共30分钟(偶发)21000并发、tm:4core 4slots 4G共有251个container总共花费了15分钟31000并发、tm:4core 10slots 4G共有251个container总共花费了15分钟分析1.从12可以看出因为我们的任务是无状态的内存的提升并没有提高运行效率且增加了yarn部署container的负担。2.1000并发下提高slots与core的比速度也没有发生变化可能是因为jm调度效率提升资源共享状况提升弥补了单线程cpu的运行效率。1.3. 任务3数据源: 在hdfs上有1000多个文件每个文件十几k压缩后的大小总共大约320万条数据处理对于每条数据通过正则去摄取目标数据。序号任务设置描述运行结果11000并发、tm:4core 4slots 4G总共有251个container部署花费了6分钟运行了5分钟21000并发、tm:4core 10slots 4G总共有251个container部署花费了6分钟运行了5分钟对于一个文件数特别多但文件都很小的情况下我这里在相同cores下提高slot数并发那在相同任务并发的情况下task的数量将减少这样会增加tm资源共享的能力一批数据运行完之后调度的时间变短,再假设每次仅通过一次拉取数据就能处理完一个文件即在调高slot数量之后处理速度还高于flink“单次”处理数据速度峰值这样总的处理时间将会和原来差不多或者更短。四、离线任务性能调优小结1. 任务稳定性Tm任务的配置不能太大10core 10slots 15G为了保证稳定调小到4core … 之前启动大于100个tm时任务就开始运行不稳定易导致HDFS集群问题或flink内部通讯问题现在运行250个后还可以稳定运行。2. 运行效率与1core多并发IO型任务的调整思路a. 当单个文件很大时9.6G/1G:43万条,需要处理速度峰值,此时需要numOfCores numOfslots,速度将达到1.2亿/minb. 当单个文件很小时十几kb不到100条一个文件这时更在意调度的效率以及Tm中资源共享的能力简单的说在意单位时间内有更多slot去处理任务。此时可以设置numOfcores n * numOfslots随着n的提高处理速度有所下降但根据文件但因为一个tm中slot的增多jm的调度能力提高。具体数据见上更细节地TCP链接的流控分析ing3. 内存调整对于没有状态来一条处理一条的离线任务内存的使用要求没有那么强之前nCores1.5nG 可以调整为 numOfcoresnumOfMemory。一方面节约资源对于1000并发的任务内存可节约0.5*1000500G的内存另一方面yarn在调度container时压力也会变小因为要去寻找更高的内存。实时任务的资源优化配置更加复杂相关参考见How To Size Your Apache Flink® Cluster: A Back-of-the-Envelope Calculation3 (More) Tips for Optimizing Apache Flink Applications
