2017年柏林Flink Forward大会上Robert Metger的"Keep It Going: How to Reiably and Efficiently Operate Apache Flink"的演讲很受欢迎。Robert的其中一个主题演讲涉及到了如何估算Flink集群规模。Flink Forward大会的观众们认为这个计算方法对他们很有用，因此我们把他的演讲主题转变成这篇博客。
Flink社区上经常被问起的一个问题是当从开发转到线上时，如何估算集群规模大小。当然，最准确的的答案是根据需要，但是这并没有什么用。这篇博客提出了一系列问题使你能够计算出一些基准。
1 通过数学建立基准
首先，思考一下你的应用操作需要的资源基准的指标。
关键指标如下：
- 每秒的数据量和每条数据的大小
- 去重key的数量和每个key的state大小
- state更新的数量和state backend的方式
最后，考虑一下你的服务等级协议(SLAS)，比如宕机时间、延迟和最大的吞吐量。这些指标将直接影响你的容量计算。
接下来，看一下基于预算可用的资源大小。比如：
- 网络容量，需要考虑外部服务的网络消耗，比如Kakfa、HDFS等。
- 磁盘带宽，比如你使用磁盘的state backend，比如RocksDB。同时需要考虑外部服务的磁盘使用，比如Kafka、HDFS等。
- 机器的数量和它们的CPU和内存。
基于上述这些因素，你现在能够估算正常流程的的资源基准。另外，还需要增加一些资源用作异常的恢复和checkpointing。
2 样例：计算
我现在通过一个集群上的虚拟job部署来描述整个资源基准的建立过程。信封背计算法的所用到的数字是不精准的，同时并没有考虑的很全面。在后面，我会指出在做计算时的忽视的一些点。
2.1 Flink流式应用样例和硬件
在这个案列中，我将部署典型的Flink流式应用，Kafka Topic的数据作为数据源。这个流接着使用keyed， aggregating window操作转换。窗口操作执行5分钟的窗口聚合。同时假设有源源不断的数据进来，window被设置成1分钟滑动一次。
这表示每分钟执行一次过去5分钟内的窗口聚合。这个流式应用根据userId字段进行聚合。Kafka Topic中消息的大小平均是2KB。
吞吐量是每秒100万条消息。为了理解窗口操作的state大小，你需要知道distinct Keys的数量，就是userIds的数量，这边大约是5000万个不同的用户ID。对于每个用户，你需要计算4个数字，通过longs(8 byte)存储。
现在，让我们总结一下这个任务的关键指标：
- 消息大小：2KB
- 吞吐量：1000000 msg/sec
- Distinct Keys: 500000000(窗口聚合：每个key4个long大小)
- Checkpointing: 每分钟一次
— 硬件：
- 5台机器
- 10 gigabit 以太网
- 每台机器运行一个Flink TaskManager
- 磁盘通过网络挂载
— Kafka 独立部署
总共有5台机器运行这个job，每台机器上运行一个TaskManager。磁盘通过网络挂载，同时有10 gigabit的以太网接入。同时Kafka是独立部署在其他机器上。
每台机器有16CPU核。为了简化的需要，这边不考虑CPU和内存的使用情况。在实际情况下，你需要根据应用逻辑和state backend的使用，来考虑内存的使用。这个例子使用RocksDB state backend。（它是健壮的，同时对内存需求比较低）。
2.2 单机计算
为了理解整个job运行部署的资源需求，最容易的方式是关注单台机器和TaskManager的操作。你可以通过单台机器计算出来的数字来推断整个集群的资源需求。
默认(所有的操作都有并行度和没有特殊的调度限制)所有的操作在每台机器上都有运行。
在这个例子中，Kafka source， 窗口操作和Kafka sink都运行在每台机器上。
keyBy是一个分离的操作，因此资源需求计算比较容易。在现实中，keyBy是一个API，连接了Kafak Source和窗口操作。
我现在将从头到底理解这些操作的网络资源需求。
2.3 Kafka source
为了计算Kafka source收到的数据量，首先需要计算Kafka的聚合输入。sources每秒收到100万消息，每条消息2KB大小。
2GB/s除以5台机器，得到如下结果：
集群中每台机器上TaskManager的source收到400MB/s的数据。
2.4 Shuffle / keyBY
接下来，你需要确保同一个key的所有事件落在一些机器上。这边，你从kafka中读取的数据可能被重新分区。
shuffle过程发送所有拥有相同key的数据到同一台机器，因此这边把400MB/s的数据分割成一个根据userId分区的流。
平均来看，你将发送80MB/s的数据到每一台机器。这个分析是从单台机器的角度，但是一些数据已经在目标机器上了，因此要减去80MB/s。
2.5 Window Emit and Kafka Sink
接下去的问题是窗口操作发送多少数据到Kafka Sink。结果是67MB/s，让我们看一下如何计算。
窗口操作为每个key保持了4个数字(longs)聚合。每一分钟，操作将发送当前的聚合值。每个key发送2ints(user_id, window_ts)和4 longs。
然后乘以keys数量(500000000除以机器数量)
然后计算每秒的大小：
这表示每个TaskManager从窗口操作中平均发送67MB/s的用户数据。因为Kafka sink运行在每个TaskManager上，所以没有进一步的分区操作。这就是从Flink到Kafka的发送的数据量。
从窗口操作中得到的数据每分钟会发送一次。在实际中，这个操作不会发以67MB/s的发送数据，而是在一分钟之内的几秒间到达最大带宽。
现在，总结一下：
- 进来的数据：720MB/s(400+320)per machine
- 出去的数据：387MB/s(320+67)per machine
2.6 State和Checkpointing
到目前为止，我们仅仅计算了Flink处理的用户数据。你同时还需要考虑磁盘的使用，比如存储state 和checkpointing。为了计算磁盘的花销，你需要查看窗口计算如何进入state。Kafka Source也需要保持一些state，但是跟窗口操作的state相比，可以忽略不计。
为了理解窗口操作的state大小，让我们换一个角度看这个问题。Flink计算5分钟的时间窗口，并且1分钟滑动一次。Flink是通过保持5个窗口来实现滑动窗口。根据先前提到的，在使用窗口时，你需要为每个窗口保持40bytes的状态，并且窗口是提前聚合的。对于每一条到来的事件，你首先需要取出当前聚合值，再更新聚合值，然后把新值写回去。
这意味着：
有40MB/s的磁盘读写（每台机器上）。根据先前说的，磁盘是通过网络挂载的。因此需要在先前的基础上增加这个值。
现在总共需要的资源如下：
- 进入的数据：760MB/s(400MB/s data in + 320MB/s shuffle + 40MB/s state)
- 出去的数据：427MB/s(320MB/s shuffle + 67MB/s data out + 40MB/s state)
上述的计算只考虑了事件到达窗口操作时触发时state的进入。此外，你还需要checkpoint和容错机制。因为，如果一台机器或者其他任何东西挂掉，你需要恢复你的窗口并继续处理。
根据先前所说，Checkpointing是每隔1分钟执行一次，并且每个checkpoint会复制整个job的状态到(通过网络挂载)文件系统。
现在，让我们快速的计算一下每台机器的state大小：
接着算每秒的值：
和窗口操作类似，checkpointing也是每分钟执行一次。它尝试全速发送数据到外部存储。Checkpointing引起了额外的state进入。（自从Flink1.3后，RocksDB支持增量checkpointing来降低每次checkpoint时所需的网络传输。）
计算更新如下：
- 进入的数据：760MB/s(400 + 320 + 40)
- 出去的数据：760MB/s(320 + 67 + 40 + 333)
这意味着整个集群网络流量是：
400是80MB的state读写乘以5台机器。2335是Kafka进和出的总值。
整个硬件的网络可用容量如下：
免责声明:上述这些计算没有包含协议的花销，比如TCP、Ethernet和RPC（在Flink、Kafka和HDFS等中）。但是上述的计算仍旧对如何计算一个job的资源有指导意义。
Scale Your Way
基于我的分析，这个例子中，5个节点的集群，在典型的操作中，每个机器需要处理760MB/s的数据进出，同时每台机器可以处理的容量是1250MB/s。这样保留了40%的网络容量来应对我刚才提到的复杂度，比如网络协议花销，事件重放，数据倾斜引起的不平均的负载等。
当然，没有一个标准答案来说明留40%的余量是否合适。但是这个算法可以给你一个好的开始。尝试上述的计算，修改上述的参数为你自己的参数。Happy scaling！
翻译源
How To Size Your Apache Flink® Cluster: A Back-of-the-Envelope Calculation
