什么网站做海宁的房产好,北控京奥建设有限公司网站,wps wordpress,海南网站推广建设分析痛点笔者线上有一个 Flink 任务消费 Kafka 数据#xff0c;将数据转换后#xff0c;在 Flink 的 Sink 算子内部调用第三方 api 将数据上报到第三方的数据分析平台。这里使用批量同步 api#xff0c;即#xff1a;每 50 条数据请求一次第三方接口#xff0c;可以通过批…分析痛点笔者线上有一个 Flink 任务消费 Kafka 数据将数据转换后在 Flink 的 Sink 算子内部调用第三方 api 将数据上报到第三方的数据分析平台。这里使用批量同步 api即每 50 条数据请求一次第三方接口可以通过批量 api 来提高请求效率。由于调用的外网接口所以每次调用 api 比较耗时。假如批次大小为 50且请求接口的平均响应时间为 50ms使用同步 api因此第一次请求响应以后才会发起第二次请求。请求示意图如下所示平均下来每 50 ms 向第三方服务器发送 50 条数据也就是每个并行度 1 秒钟处理 1000 条数据。假设当前业务数据量为每秒 10 万条数据那么 Flink Sink 算子的并行度需要设置为 100 才能正常处理线上数据。从 Flink 资源分配来讲100 个并行度需要申请 100 颗 CPU因此当前 Flink 任务需要占用集群中 100 颗 CPU 以及不少的内存资源。请问此时 Flink Sink 算子的 CPU 或者内存压力大吗上述请求示意图可以看出 Flink 任务发出请求到响应这 50ms 期间Flink Sink 算子只是在 wait并没有实质性的工作。因此CPU 使用率肯定很低当前任务的瓶颈明显在网络 IO。最后结论是 Flink 任务申请了 100 颗 CPU导致 yarn 或其他资源调度框架没有资源了但是这 100 颗 CPU 的使用率并不高这里能不能优化通过提高 CPU 的使用率从而少申请一些 CPU 呢同步批量请求优化为异步请求首先可以想到的是将同步请求改为异步请求使得任务不会阻塞在网络请求这一环节请求示意图如下所示。异步请求相比同步请求而言优化点在于每次发出请求时不需要等待请求响应后再发送下一次请求而是当下一批次的 50 条数据准备好之后直接向第三方服务器发送请求。每次发送请求后Flink Sink 算子的客户端需要注册监听器来等待响应当响应失败时需要做重试或者回滚策略。通过异步请求的方式可以优化网络瓶颈假如 Flink Sink 算子的单个并行度平均 10ms 接收到 50 条数据那么使用异步 api 的方式平均 1 秒可以处理 5000 条数据整个 Flink 任务的性能提高了 5 倍。对于每秒 10 万数据量的业务这里仅需要申请 20 颗 CPU 资源即可。关于异步 api 的具体使用可以根据场景具体设计这里不详细讨论。多线程 Client 模式对于一些不支持异步 api 的场景可能并不能使用上述优化方案同样为了提高 CPU 使用率可以在 Flink Sink 端使用多线程的方案。如下图所示可以在 Flink Sink 端开启 5 个请求第三方服务器的 Client 线程Client1、Client2、Client3、Client4、Client5。这五个线程内分别使用同步批量请求的 Client单个 Client 还是保持 50 条记录为一个批次即 50 条记录请求一次第三方 api。请求第三方 api 耗时主要在于网络 IO性能瓶颈在于网络请求延迟因此如果变成 5 个 Client 线程每个 Client 的单次请求平均耗时还能保持在 50ms除非网络请求已经达到了带宽上限或整个任务又遇到其他瓶颈。所以多线程模式下使用同步批量 api 也能将请求效率提升 5 倍。说明多线程的方案不仅限于请求第三方接口对于非 CPU 密集型的任务也可以使用该方案在降低 CPU 数量的同时单个 CPU 承担多个线程的工作从而提高 CPU 利用率。例如请求 HBase 的任务或磁盘 IO 是瓶颈的任务可以降低任务的并行度使得每个并行度内处理多个线程。Flink 算子内多线程实现Sink 算子的单个并行度内现在有 5 个 Client 用于消费数据但 Sink 算子的数据都来自于上游算子。如下图所示一个简单的实现方式是 Sink 算子接收到上游数据后通过轮循或随机的策略将数据分发给 5 个 Client 线程。但是轮循或者随机策略会存在问题假如 5 个 Client 中 Client3 线程消费较慢会导致给 Client3 分发数据时被阻塞从而使得其他正常消费的线程 Client1、2、4、5 也被分发不到数据。为了解决上述问题可以在 Sink 算子内申请一个数据缓冲队列队列有先进先出FIFO的特性。Sink 算子接收到的数据直接插入到队列尾部五个 Client 线程不断地从队首取数据并消费即Sink 算子先接收的数据 Client 先消费后接收的数据 Client 后消费。若队列一直是满的说明 Client 线程消费较慢、Sink 算子上游生产数据较快。若队列一直为空说明 Client 线程消费较快、Sink 算子的上游生产数据较慢。五个线程共用同一个队列完美地解决了单个线程消费慢的问题当 Client3 线程阻塞时不影响其他线程从队列中消费数据。这里使用队列还起到了削峰填谷的作用。代码实现原理明白了具体代码如下所示首先是消费数据的 Client 线程代码代码逻辑很简单一直从 bufferQueue 中 poll 数据取出数据后执行相应的消费逻辑即可在本案例中消费逻辑便是 Client 积攒批次并调用第三方 api。public class MultiThreadConsumerClient implements Runnable {private LinkedBlockingQueueString bufferQueue;public MultiThreadConsumerClient(LinkedBlockingQueueString bufferQueue) {this.bufferQueue bufferQueue;}Overridepublic void run() {String entity;while (true){// 从 bufferQueue 的队首消费数据entity bufferQueue.poll();// 执行 client 消费数据的逻辑doSomething(entity);}}// client 消费数据的逻辑private void doSomething(String entity) {// client 积攒批次并调用第三方 api}
}Sink 算子代码如下所示在 open 方法中需要初始化线程池、数据缓冲队列并创建开启消费者线程在 invoke 方法中只需要往 bufferQueue 的队尾添加数据即可。public class MultiThreadConsumerSink extends RichSinkFunctionString {// Client 线程的默认数量private final int DEFAULT_CLIENT_THREAD_NUM 5;// 数据缓冲队列的默认容量private final int DEFAULT_QUEUE_CAPACITY 5000;private LinkedBlockingQueueString bufferQueue;Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);// new 一个容量为 DEFAULT_CLIENT_THREAD_NUM 的线程池ThreadPoolExecutor threadPoolExecutor new ThreadPoolExecutor(DEFAULT_CLIENT_THREAD_NUM, DEFAULT_CLIENT_THREAD_NUM,0L,TimeUnit.MILLISECONDS, new LinkedBlockingQueue());// new 一个容量为 DEFAULT_QUEUE_CAPACITY 的数据缓冲队列this.bufferQueue Queues.newLinkedBlockingQueue(DEFAULT_QUEUE_CAPACITY);// 创建并开启消费者线程MultiThreadConsumerClient consumerClient new MultiThreadConsumerClient(bufferQueue);for (int i0; i DEFAULT_CLIENT_THREAD_NUM; i) {threadPoolExecutor.execute(consumerClient);}}Overridepublic void invoke(String value, Context context) throws Exception {// 往 bufferQueue 的队尾添加数据bufferQueue.put(value);}
}代码逻辑相对比较简单请问上述 Sink 能保证 Exactly Once 吗答不能保证 Exactly OnceFlink 要想端对端保证 Exactly Once必须要求外部组件支持事务这里第三方接口明显不支持事务。那么上述 Sink 能保证 At Lease Once 吗言外之意上述 Sink 会丢数据吗答会丢数据。因为上述案例中使用的批量 api 来消费数据假如批量 api 是每积攒 50 条数据请求一次第三方接口当做 Checkpoint 时可能只积攒了 30 条数据所以做 Checkpoint 时内存中可能还有数据未发送到外部系统。而且数据缓冲队列中可能还有缓存的数据因此上述 Sink 在做 Checkpoint 时会出现 Checkpoint 之前的数据未完全消费的情况。例如Flink 任务消费的 Kafka 数据当做 Checkpoint 时Flink 任务消费到 offset 为 10000 的位置但实际上 offset 10000 之前的一小部分数据可能还在数据缓冲队列中尚未完全消费或者因为没积攒够一定批次所以数据缓存在 client 中并未请求到第三方。当任务失败后Flink 任务从 Checkpoint 处恢复会从 offset 为 10000 的位置开始消费此时 offset 10000 之前的一小部分缓存在内存缓冲队列中的数据不会再被消费于是就出现了丢数据情况。处理丢数据情况如何保证数据不丢失呢很简单可以在 Checkpoint 时强制将数据缓冲区的数据全部消费完并对 client 执行 flush 操作保证 client 端不会缓存数据。实现思路Sink 算子可以实现 CheckpointedFunction 接口当做 Checkpoint 时会调用 snapshotState 方法方法内可以触发 client 的 flush 操作。但 client 在 MultiThreadConsumerClient 对应的五个线程中需要考虑线程同步的问题即Sink 算子的 snapshotState 方法中做一个操作要使得五个 Client 线程感知到当前正在执行 Checkpoint此时应该把数据缓冲区的数据全部消费完并对 client 执行过 flush 操作。如何实现呢需要借助 CyclicBarrier。CyclicBarrier 会让所有线程都等待某个操作完成后才会继续下一步行动。在这里可以使用 CyclicBarrier让 Checkpoint 等待所有的 client 将数据缓冲区的数据全部消费完并对 client 执行过 flush 操作言外之意offset 10000 之前的数据必须全部消费完成才允许 Checkpoint 执行完成。这样就可以保证 Checkpoint 时不会有数据被缓存在内存可以保证数据源 offset 10000 之前的数据都消费完成。MultiThreadConsumerSink 具体代码如下所示public class MultiThreadConsumerSink extends RichSinkFunctionString {// Client 线程的默认数量private final int DEFAULT_CLIENT_THREAD_NUM 5;// 数据缓冲队列的默认容量private final int DEFAULT_QUEUE_CAPACITY 5000;private LinkedBlockingQueueString bufferQueue;Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);// new 一个容量为 DEFAULT_CLIENT_THREAD_NUM 的线程池ThreadPoolExecutor threadPoolExecutor new ThreadPoolExecutor(DEFAULT_CLIENT_THREAD_NUM, DEFAULT_CLIENT_THREAD_NUM,0L,TimeUnit.MILLISECONDS, new LinkedBlockingQueue());// new 一个容量为 DEFAULT_QUEUE_CAPACITY 的数据缓冲队列this.bufferQueue Queues.newLinkedBlockingQueue(DEFAULT_QUEUE_CAPACITY);// 创建并开启消费者线程MultiThreadConsumerClient consumerClient new MultiThreadConsumerClient(bufferQueue);for (int i0; i DEFAULT_CLIENT_THREAD_NUM; i) {threadPoolExecutor.execute(consumerClient);}}Overridepublic void invoke(String value, Context context) throws Exception {// 往 bufferQueue 的队尾添加数据bufferQueue.put(value);}
}MultiThreadConsumerSink 实现了 CheckpointedFunction 接口在 open 方法中增加了 CyclicBarrier 的初始化CyclicBarrier 预期容量设置为 client 线程数加一表示当 client 线程数加一个线程都执行了 await 操作时所有的线程的 await 方法才会执行完成。这里为什么要加一呢因为除了 client 线程外 snapshotState 方法中也需要执行过 await。当做 Checkpoint 时 snapshotState 方法中执行 clientBarrier.await()等待所有的 client 线程将缓冲区数据消费完。snapshotState 方法执行过程中 invoke 方法不会被执行即Checkpoint 过程中数据缓冲队列不会增加数据所以 client 线程很快就可以将缓冲队列中的数据消费完。MultiThreadConsumerClient 具体代码如下所示public class MultiThreadConsumerSink extends RichSinkFunctionString implements CheckpointedFunction {private Logger LOG LoggerFactory.getLogger(MultiThreadConsumerSink.class);// Client 线程的默认数量private final int DEFAULT_CLIENT_THREAD_NUM 5;// 数据缓冲队列的默认容量private final int DEFAULT_QUEUE_CAPACITY 5000;private LinkedBlockingQueueString bufferQueue;private CyclicBarrier clientBarrier;Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);// new 一个容量为 DEFAULT_CLIENT_THREAD_NUM 的线程池ThreadPoolExecutor threadPoolExecutor new ThreadPoolExecutor(DEFAULT_CLIENT_THREAD_NUM, DEFAULT_CLIENT_THREAD_NUM,0L,TimeUnit.MILLISECONDS, new LinkedBlockingQueue());// new 一个容量为 DEFAULT_QUEUE_CAPACITY 的数据缓冲队列this.bufferQueue Queues.newLinkedBlockingQueue(DEFAULT_QUEUE_CAPACITY);// barrier 需要拦截 (DEFAULT_CLIENT_THREAD_NUM 1) 个线程this.clientBarrier new CyclicBarrier(DEFAULT_CLIENT_THREAD_NUM 1);// 创建并开启消费者线程MultiThreadConsumerClient consumerClient new MultiThreadConsumerClient(bufferQueue, clientBarrier);for (int i0; i DEFAULT_CLIENT_THREAD_NUM; i) {threadPoolExecutor.execute(consumerClient);}}Overridepublic void invoke(String value, Context context) throws Exception {// 往 bufferQueue 的队尾添加数据bufferQueue.put(value);}Overridepublic void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {LOG.info(snapshotState : 所有的 client 准备 flush !!!);// barrier 开始等待clientBarrier.await();}Overridepublic void initializeState(FunctionInitializationContext functionInitializationContext) throws Exception {}}从数据缓冲队列中 poll 数据时增加了 timeout 时间为 50ms。如果从队列中拿到数据则执行消费数据的逻辑若拿不到数据说明数据缓冲队列中数据消费完了。此时需要判断是否有等待的 CyclicBarrier如果有等待的 CyclicBarrier 说明此时正在执行 Checkpoint所以 client 需要执行 flush 操作。flush 完成后Client 线程执行 barrier.await() 操作。当所有的 Client 线程都执行到 await 时所有的 barrier.await() 都会被执行完。此时 Sink 算子的 snapshotState 方法就会执行完。通过这种策略可以保证 Checkpoint 时将数据缓冲区中的数据消费完client 执行 flush 操作可以保证 client 端不会缓存数据。总结分析到这里我们设计的 Sink 终于可以保证不丢失数据了。对 CyclicBarrier 不了解的同学请 Google 或百度查询。再次强调这里多线程的方案不仅限于请求第三方接口对于非 CPU 密集型的任务都可以使用该方案来提高 CPU 利用率且该方案不仅限于 Sink 算子各种算子都适用。本文主要希望帮助大家理解 Flink 中使用多线程的优化及在 Flink 算子中使用多线程如何保证不丢数据。原文链接本文为阿里云内容未经允许不得转载
