在 Flink 算子中使用多线程如何保证不丢数据?

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 本人通过分析痛点、同步批量请求优化为异步请求、多线程 Client 模式、Flink 算子内多线程实现以及总结四部分帮助大家理解 Flink 中使用多线程的优化及在 Flink 算子中使用多线程如何保证不丢数据。

分析痛点

笔者线上有一个 Flink 任务消费 Kafka 数据,将数据转换后,在 Flink 的 Sink 算子内部调用第三方 api 将数据上报到第三方的数据分析平台。这里使用批量同步 api,即:每 50 条数据请求一次第三方接口,可以通过批量 api 来提高请求效率。由于调用的外网接口,所以每次调用 api 比较耗时。假如批次大小为 50,且请求接口的平均响应时间为 50ms,使用同步 api,因此第一次请求响应以后才会发起第二次请求。请求示意图如下所示:

1.jpg

平均下来,每 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 呢?

同步批量请求优化为异步请求

首先可以想到的是将同步请求改为异步请求,使得任务不会阻塞在网络请求这一环节,请求示意图如下所示。

2.jpg

异步请求相比同步请求而言,优化点在于每次发出请求时,不需要等待请求响应后再发送下一次请求,而是当下一批次的 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 倍。

3.jpg

说明:多线程的方案,不仅限于请求第三方接口,对于非 CPU 密集型的任务也可以使用该方案,在降低 CPU 数量的同时,单个 CPU 承担多个线程的工作,从而提高 CPU 利用率。例如:请求 HBase 的任务或磁盘 IO 是瓶颈的任务,可以降低任务的并行度,使得每个并行度内处理多个线程。

Flink 算子内多线程实现

Sink 算子的单个并行度内现在有 5 个 Client 用于消费数据,但 Sink 算子的数据都来自于上游算子。如下图所示,一个简单的实现方式是 Sink 算子接收到上游数据后通过轮循或随机的策略将数据分发给 5 个 Client 线程。

4.jpg

但是轮循或者随机策略会存在问题,假如 5 个 Client 中 Client3 线程消费较慢,会导致给 Client3 分发数据时被阻塞,从而使得其他正常消费的线程 Client1、2、4、5 也被分发不到数据。

为了解决上述问题,可以在 Sink 算子内申请一个数据缓冲队列,队列有先进先出(FIFO)的特性。Sink 算子接收到的数据直接插入到队列尾部,五个 Client 线程不断地从队首取数据并消费,即:Sink 算子先接收的数据 Client 先消费,后接收的数据 Client 后消费。

  • 若队列一直是满的,说明 Client 线程消费较慢、Sink 算子上游生产数据较快。
  • 若队列一直为空,说明 Client 线程消费较快、Sink 算子的上游生产数据较慢。

五个线程共用同一个队列完美地解决了单个线程消费慢的问题,当 Client3 线程阻塞时,不影响其他线程从队列中消费数据。这里使用队列还起到了削峰填谷的作用。

5.jpg

代码实现

原理明白了,具体代码如下所示,首先是消费数据的 Client 线程代码,代码逻辑很简单,一直从 bufferQueue 中 poll 数据,取出数据后,执行相应的消费逻辑即可,在本案例中消费逻辑便是 Client 积攒批次并调用第三方 api。

public class MultiThreadConsumerClient implements Runnable {

    private LinkedBlockingQueue<String> bufferQueue;

    public MultiThreadConsumerClient(LinkedBlockingQueue<String> bufferQueue) {
        this.bufferQueue = bufferQueue;
    }

    @Override
    public 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 RichSinkFunction<String> {
    // Client 线程的默认数量
    private final int DEFAULT_CLIENT_THREAD_NUM = 5;
    // 数据缓冲队列的默认容量
    private final int DEFAULT_QUEUE_CAPACITY = 5000;

    private LinkedBlockingQueue<String> bufferQueue;

    @Override
    public 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 i=0; i < DEFAULT_CLIENT_THREAD_NUM; i++) {
            threadPoolExecutor.execute(consumerClient);
        }
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 往 bufferQueue 的队尾添加数据
        bufferQueue.put(value);
    }
}

代码逻辑相对比较简单,请问上述 Sink 能保证 Exactly Once 吗?

答:不能保证 Exactly Once,Flink 要想端对端保证 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 之前的一小部分缓存在内存缓冲队列中的数据不会再被消费,于是就出现了丢数据情况。

6.jpg

处理丢数据情况

如何保证数据不丢失呢?很简单,可以在 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 RichSinkFunction<String> {
    // Client 线程的默认数量
    private final int DEFAULT_CLIENT_THREAD_NUM = 5;
    // 数据缓冲队列的默认容量
    private final int DEFAULT_QUEUE_CAPACITY = 5000;

    private LinkedBlockingQueue<String> bufferQueue;

    @Override
    public 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 i=0; i < DEFAULT_CLIENT_THREAD_NUM; i++) {
            threadPoolExecutor.execute(consumerClient);
        }
    }

    @Override
    public 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 RichSinkFunction<String> 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 LinkedBlockingQueue<String> bufferQueue;
    private CyclicBarrier clientBarrier;

    @Override
    public 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 i=0; i < DEFAULT_CLIENT_THREAD_NUM; i++) {
            threadPoolExecutor.execute(consumerClient);
        }
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 往 bufferQueue 的队尾添加数据
        bufferQueue.put(value);
    }

    @Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        LOG.info("snapshotState : 所有的 client 准备 flush !!!");
        // barrier 开始等待
        clientBarrier.await();
    }

    @Override
    public 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 算子中使用多线程如何保证不丢数据。

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
5月前
|
存储 消息中间件 Kafka
基于 Flink 的中国电信星海时空数据多引擎实时改造
本文整理自中国电信集团大数据架构师李新虎老师在Flink Forward Asia 2024的分享,围绕星海时空智能系统展开,涵盖四个核心部分:时空数据现状、实时场景多引擎化、典型应用及未来展望。系统日处理8000亿条数据,具备亚米级定位能力,通过Flink多引擎架构解决数据膨胀与响应时效等问题,优化资源利用并提升计算效率。应用场景包括运动状态识别、个体行为分析和群智感知,未来将推进湖仓一体改造与三维时空服务体系建设,助力数字化转型与智慧城市建设。
583 3
基于 Flink 的中国电信星海时空数据多引擎实时改造
|
2月前
|
存储 消息中间件 搜索推荐
京东零售基于Flink的推荐系统智能数据体系
摘要:本文整理自京东零售技术专家张颖老师,在 Flink Forward Asia 2024 生产实践(二)专场中的分享,介绍了基于Flink构建的推荐系统数据,以及Flink智能体系带来的智能服务功能。内容分为以下六个部分: 推荐系统架构 索引 样本 特征 可解释 指标 Tips:关注「公众号」回复 FFA 2024 查看会后资料~
178 1
京东零售基于Flink的推荐系统智能数据体系
|
6月前
|
Oracle 关系型数据库 Java
【YashanDB知识库】Flink CDC实时同步Oracle数据到崖山
本文介绍通过Flink CDC实现Oracle数据实时同步至崖山数据库(YashanDB)的方法,支持全量与增量同步,并涵盖新增、修改和删除的DML操作。内容包括环境准备(如JDK、Flink版本等)、Oracle日志归档启用、用户权限配置、增量日志记录设置、元数据迁移、Flink安装与配置、生成Flink SQL文件、Streampark部署,以及创建和启动实时同步任务的具体步骤。适合需要跨数据库实时同步方案的技术人员参考。
【YashanDB知识库】Flink CDC实时同步Oracle数据到崖山
|
7月前
|
Java 关系型数据库 MySQL
SpringBoot 通过集成 Flink CDC 来实时追踪 MySql 数据变动
通过详细的步骤和示例代码,您可以在 SpringBoot 项目中成功集成 Flink CDC,并实时追踪 MySQL 数据库的变动。
1584 45
|
6月前
|
消息中间件 关系型数据库 Kafka
阿里云基于 Flink CDC 的现代数据栈云上实践
阿里云基于 Flink CDC 的现代数据栈云上实践
|
10月前
|
存储 监控 数据处理
flink 向doris 数据库写入数据时出现背压如何排查?
本文介绍了如何确定和解决Flink任务向Doris数据库写入数据时遇到的背压问题。首先通过Flink Web UI和性能指标监控识别背压,然后从Doris数据库性能、网络连接稳定性、Flink任务数据处理逻辑及资源配置等方面排查原因,并通过分析相关日志进一步定位问题。
688 61
|
11月前
|
运维 数据处理 Apache
数据实时计算产品对比测评报告:阿里云实时计算Flink版
数据实时计算产品对比测评报告:阿里云实时计算Flink版
|
11月前
|
分布式计算 监控 大数据
大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu
大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu
212 1
|
11月前
|
消息中间件 Java Kafka
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
243 1
|
12月前
|
SQL 安全 数据处理
揭秘数据脱敏神器:Flink SQL的神秘力量,守护你的数据宝藏!
【9月更文挑战第7天】在大数据时代,数据管理和处理尤为重要,尤其在保障数据安全与隐私方面。本文探讨如何利用Flink SQL实现数据脱敏,为实时数据处理提供有效的隐私保护方案。数据脱敏涉及在处理、存储或传输前对敏感数据进行加密、遮蔽或替换,以遵守数据保护法规(如GDPR)。Flink SQL通过内置函数和表达式支持这一过程。
178 2

相关产品

  • 实时计算 Flink版