前言
本文主要译自Flink Forward 2017的柏林站中Robert Metzger的有关集群规划的How to size your flink cluster一文。该文中主要是考虑网络资源,博主结合自己的使用经验对文中省略的做了一定补充,同时也非常欢迎大伙留言补充。
本文非直译,原文链接如下:https://www.ververica.com/blog/how-to-size-your-apache-flink-cluster-general-guidelines
文中拿捏不准的地方,均附有英文原文。若是有表述不合适的,欢迎大伙留言指出。
1、关键参数与资源
为估算Flink集群所需资源,首先我们需要根据Flink任务中的指标给出集群的最低资源需求(baseline)。
1.1 指标(metric):
1)每秒的record数和每个record的大小;
2)不同key的个数和每个key产生state的大小;
3)state的更新方式以及state的访问模式
此外还需考虑SLA(服务级别协议)。例如,可能愿意接受的停机时间,可接受的延迟或最大吞吐量,因为此类SLA会对Flink群集的大小产生影响。
1.2 资源
在给Flink集群做规划时,我们需要考虑集群的资源,但这里的资源一般指什么呢?一般有以下几种:
1)网络容量。在考虑网络容量时,我们也需要考虑到可能使用网络的其他服务,如Kafka、HDFS等;
2)磁盘带宽。当我们的容错机制是基于磁盘的,如RockDB、HDFS,此时也有可能需要考虑到Kafka,因为其也是将数据存在磁盘的;
3)节点数量以及能提供的CPU和内存;
2、例子
Flink例子的拓扑图1如下:
该例子从kafka消费message,以用户id(userId)做keyBy后,经过window算子聚合(window算子为sliding window,其窗口大小为5min,间隔是1min),处理后的消息写入到kafka中。
2.1 任务metrics
从kafka消费的record平均大小为2KB,吞吐量为1百万/s,userId的个数为5亿(5*10^9)。该任务的关键指标(key metric)如下:
2.2 硬件
1)5个节点,每个节点有一个TaskManager;2)万兆网;3)磁盘通过网络连接(本例中集群部署在云上,物理机得另外考虑);此外,kafka是单独的集群。如下图2:
每个节点是16核,为简化,文中暂不考虑CUP和内存的需求。在实际的生产中需要根据任务逻辑和容错方式去考虑内存。本例的状态是通过RockDB的方式存储,该方式对内存的要求较小。
2.3 单节点资源需求
为方便分析,我们先考虑单节点上的资源需求,集群整体的需求可以大致通过乘以节点数得到。例子中,每个算子的并行度相同且没有其他特殊调度限制,每个节点拥有流任务的所有算子,即每个节点上都有Kafka source、window、Kafka sink算子,如下图3:
为方便计算资源,上图中KeyBy算子单独给出,但在实际中KeyBy是Kafka算子和window算子之间链接的配置属性。下面将结合图3从上往下分析网络资源的需求(network resource requirement)。
2.3.1 Kafka Source
为计算从单个Kafka Source的拿到的数据,我们先计算从Kafka拿到数据的综合,计算过程如下:
1)每秒1,000,000条,每条大小为2KB ,每秒获得总数据为:
2KB×1,000,000/s=2GB/s
2)Flink集群中每个节点每秒获得数据为
2GB/s÷5=400MB/s
2.3.2 Shuffle过程(KeyBy)
经过KeyBy后,具有相同userId的数据将会在一个节点上,但是Kafka可能根据不同的元数据进行分区(partitioned according to a different partitioning scheme),因此对一个key(userId),单个节点直接从Kafka得到的数据为400MB/s÷5=80MB/s,这样就有320MB/s的需要通过shuffle获得。
2.3.3 window emit和Kafka sink
window会发送多少数据,有多少数据会到Kafka sink?分析如下:
window算子为每个key(userId)聚合生成4个long数,每分钟发射一次,这样window每分钟为每个key会发射2个int字段(userId、window_ts)和4个long字段,总的数据量如下:
(2 x 4 bytes) + (4 x 8 bytes) = 40 bytes per key
这样5个节点,每个节点的数据量为:
500,000,000 keys x 40 bytes÷5 = 4GB
每秒的数据量为4GB/min ÷ 60 = 67MB/s,因为每个节点上都有Kafka sink,不需要额外的重分区,因此从Flink到Kafka的数据为67MB/s。在实际中,算子不会以67MB/s的恒定速度发送数据,而是每分钟最大限度地利用可用带宽几秒钟。
单节点数据总流向总结如下:
- Data in: 720MB/s (400 + 320) per machine
- Data out: 387MB/s (320 + 67) per machine
整个过程可以总结如下:
2.3.4 获取state和checkpointing
到目前为止,我们只考虑Flink处理的数据。实际上,还需考虑到state存储和checkpoint过程中所需要的网络资源。
1)state消耗的网络带宽
为弄清window算子的state大小,我们需要从另外一个角度去分析该问题。Flink的计算窗口大小为5min,滑动尺度为1min,为此Flink通过维持五个窗口实现“滑动窗口”。如在2.3.3节中提到,每个窗口Flink需要维持40Bytes的数据。每当一个event到达时,Flink将会从已有state中获得数据(40Bytes)去更新聚合值,然后将更新后的数据写入state(磁盘),如下图:
这意味每个节点将会产生40MB/s的网络消耗,计算方式如下:
40 bytes of state x 5 windows x 200,000 msg/s per machine = 40MB/s
正如文中开始提及的,磁盘是通过网络连接的,所以state读取产生的网络消耗也得考虑进去,则单节点整体的网络资源情况如下:
2)checkpoint过程
每当有新event到来上述state过程就会被触发,有时间我们为了保证当任务失败后可以恢复会开启checkpoint,本例中checkpoint设置为每隔一分钟周期性触发,每个checkpoint过程会将现有的state通过网络拷贝到系统中。每个节点一次checkpoint会拷贝的数据为:
40bytes of state x 5 windows x 100,000,000 keys = 20GB
每秒中的数据为20GB ÷ 60 = 333 MB/s。当然checkpoint过程数据同样不是以稳定的速率发送到系统中,而是会以最大的速率发送。此外,从Flink1.3以后,基于RockDB是可以实现增量checkpoint,本例暂时不考虑该特性。单节点整个任务过程网络消耗如下:
集群整体网络消耗如下:
760 + 760 x 5 + (40×2)×5 + (400+67)×5 = 10335 MB/s
(40×2)×5是5个节点state的读写过程消耗,(400+67)×5是从Kafka读和写过程消耗的(kafka数据会落盘)。
该数据仅为上述硬件设置中的可用网络容量的一半以上,如下图。
2.3.5 总结
该例子中,每个节点流进和流出的数据为760MB/s,仅为节点容量的60%(每个节点为1250MB/s),剩下的40%可以用来应对突发的情况,如网络开销、checkpoint恢复期间的数据重放或者由于数据倾斜导致的节点之间数据shuffle过大的情况等。
3、其他建议
1)CUP个数,Flink官网给出的建议是和slot的个数成比例,从而也就和任务的并行度有关了,换句话说,在考虑任务的并行度时要结合CPU的个数考虑;
2)尽量申请多的内存,内存的最小和可以通过在测试集群中测试后,大致成比例的放大到生成集群中;
3)考虑I/O,数据盘最好和日志盘分离;
4)还有其他如JobManager最好和TaskManager节点分离等;
原文地址:https://www.cnblogs.com/love-yh/p/11939023.html