导语：Apache Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。本文主要从实际案例入手并结合作者的实践经验，向各位读者分享当应用场景中异构集群无法做到负载均衡时，如何通过Flink的自定义多线程来实现异构集群的动态负载均衡。
● 1. 前言
● 2. 出现的问题与解决方案
● 2.1 出现问题
● 2.2 分析思路
● 2.3 解决方案
● 3. 技术架构
● 4. 建设成效
● 5. 结语
前言
在实时计算应用场景中经常会有对异构集群的实时调用需求，而当异构集群的服务由于机器配置、节点负载等原因无法做到负载均衡时，可以通过Flink的自定义多线程来实现对异构集群的动态负载均衡。
下面举个例子：
文本内容鉴别、图片内容鉴别、图片OCR等特征生产需求，都需要和基于GPU部署的异构集群来交互。如果GPU集群机器配置无法统一，那么就会产生负载不均的情况。即：一个GPU集群中某些节点处理的快，某些节点处理的慢，处理慢的节点往往会导致大量的超时异常，从而引起整个作业的反压。
其流程图如下：
我们借助Flink分布式的先天优势，在任务中通过Thrift RPC调用模型服务，实时获取结果后再写到特征工程，以此来构建特征生成整个链路。
出现的问题与解决方案
出现问题：
在模型服务中部署的若干个节点，节点和节点之间是完全独立的，每个节点上线/下线后都会将自己的状态更新到ZooKeeper。在Flink任务的每个subTask都会注册一个Watch，以便获取最新且可用的所有节点。对于流入到subTask的每条数据，都需要选择一个节点来完成数据的推理。
在前期，我们使用Random策略来选择节点，但是在使用过程中我们发现，如果服务端的一个模型节点性能变低，随之而来的就是数据推理的耗时变长，那么最终可能会导致Flink任务反压。而在服务端来看，很多模型节点并没有满负荷运行，但客户端反应出来的服务端性能却不够，处理完成的总QPS很低。
另外，在和模型服务通信的时候，我们采用的是同步策略，这对于一些推理耗时长、QPS高的任务来说，需要足够大的并发度才能完成数据请求。然而，这些任务的资源利用率较低，这也是生产环境的一大痛点。
分析思路：
基于上图，我们作出如下分析：
在理想情况下，我们假设服务端有节点1、节点2、节点3，且3个节点性能都一致，每个节点都开32个并行度处理。假设每条数据处理耗时都是800ms，那么每个节点的处理能力应该是40条/s，三个节点满载的情况下处理能力应该是120条/s。
而在实际生产环境下，服务端部署的机器各个节点之间处理能力是有差异的，导致其出现差异的原因主要有三点。
■ GPU物理机器有多种规格，性能之间差别较大，在进行部署时很难确保将节点部署在同一批机器上；
■ 一台机器上混部多种模型服务，相互之间会有影响；
■ 部分节点所在机器网络、磁盘等出现故障也会导致出现差异。
例如节点1和节点2部署在高性能机器上，节点并行度是32，单条数据处理耗时是800ms。节点3部署在低性能机器上，节点并行度是32，单条数据处理耗时2400ms，那么节点3的处理能力可以看作为13.3条/s。同样采用随机选择节点的策略，倘若一秒总共发送了40条数据，对于节点3来说就已经达到性能瓶颈了。假设这个时候有更多的数据选择到了节点3进行处理，就只能在服务端的队列进行排队，而如果这个队列满了就会拒绝连接。
随着任务的运行，节点3的等待队列数据会越来越多，客户端从发送请求到返回结果的耗时也会随之越来越大。这时一旦有subTask选择了这个节点，那么这个subTask就需要等待比较长的时间来完成这次的请求。在这个过程中，如果上游数据还在源源不断的流入，那么就会造成subTask的InputChannel慢慢被耗尽，随后公共Buffer Pool的空间也会被占满，进而导致subTask2卡死。又因为上游算子使用了Rebalance，最终会把整个Flink任务卡死。
这是一个典型的木桶效应，在实际应用场景中，一旦有个别节点性能差或者出现故障，就会影响整个任务的稳定性。
解决方案:
在分析事故出现的原因后，我们提出给每个节点配置一个权重，模型节点定期将权重数值上报到ZooKeeper，客户端再通过权重来给每个节点分配相应的流量。这个想法很好，也在实践过程中取得了一定的效果，但是它有一个小问题：这样做会导致每个节点的流量一直抖动，抖动的频次和上报权重的时间呈正相关的关系，而且从监控来看，这样做也会导致处理的总数据量不太稳定。
在问题初步得到缓解后，我们开始思考是不是还有其他更好的方式可以解决这个事情。已知模型节点慢了会卡死一个subTask，而这个subTask本质是一个Slot，也就是一个线程，那么我们是不是可以借用多线程的方式来解决这个问题？带着这个思考，我们又尝试了另外两种解决方式：Async I/O和多线程方案。
■ Async I/O方案
Flink 在1.2版本中引入Async I/O，其主要目的是为了解决与外部系统交互时网络延迟成为系统瓶颈的问题。通过暴露出来的API，我们可以设置最大的操作数，简单理解为Slot中异步请求的最大并发数量。测试的时候我们准备了三个节点，一个节点处理耗时2000ms，另外两个节点处理耗时500ms。
任务在刚启动的几分钟内运行还比较正常，之后处理速度较慢节点的服务端队列堆积长度开始变得越来越大，最后稳定在150左右，与此同时超时失败率也开始随之升高。我们可以看到，通过这种方式虽然可以很好的解决因为快慢节点导致反压卡死和资源利用率问题，但是仍然无法解决流量分配的问题。
■ 多线程方案
我们在每个Slot内部实现一个生产者-消费者模型，再创建和模型节点相同数量的线程，让每个线程固定的请求一个节点，就算这个节点卡死或者处理速度缓慢，也只会影响当前线程，对整个subTask的影响有限。
如上图所示，一个Slot内包含多个线程，个别线程对应的Service节点出现问题不会影响其它线程的消费。通过这种方式可以实现自适应的流量分发策略，每个线程对应一个服务端Pod，这种线程自适应阻塞的方式可以实现慢的节点少消费，快的节点多消费的目的。
在Flink以Slot为最小资源粒度情况下再进行细化，从Slot中开启若干个线程能够增大并发度，从而减少整体Slot数量，在降低资源的同时提高资源利用率。
技术架构
在Flink任务使用多线程的方案来解决RPC通信负载均衡的问题，编程模型需要做相应的改造，具体改造如下图：
建设成效
通过前后数据对比，我们发现在采用多线程方案后，效果还是很明显的。通过下图所示的服务端处理耗时这个指标我们可以看出以125.172结尾的节点处理耗时在1.7s左右，在流量分发数量指标可以看到分配给它的流量在5条/s左右；160.25的节点处理耗时在0.14s左右，分配的流量在58条/s左右，整体符合预期。
同时，为了保证整个服务的稳定性，我们增加了一些监控指标，如：缓存队列长度、模型失败率、链路耗时、写特征失败率等。
结语
本文主要介绍了Flink在异构集群调用的时候，如果出现服务端无法分发流量的情况，在客户端可以通过多线程的方式实现流量的动态负载均衡，这样做可以帮助服务端兼容高低配机型，提升机器利用效能。不过，需要特别指出的是本文中使用的多线程算子是Stateless(无状态)的，对于有状态的算子还需要酌情考虑。另外，如果服务端是可以自主分配节点的组件时，可以选择使用Async I/O方案。
