Flink + Iceberg，百亿级实时数据入湖实战

Posted 2021-08-05 zhisheng_blog

摘要：本文整理自腾讯数据湖研发高级工程师陈俊杰在 4 月 17 日 上海站 Flink Meetup 分享的《百亿级实时数据入湖实战》。内容包括：

1. 腾讯数据湖介绍
   

2. 百亿级数据场景落地
   

3. 未来规划

4. 总结
   

Tips：点击文末「阅读原文」即可查看更多技术干货～

 GitHub 地址 

欢迎大家给 Flink 点赞送 star~

一、腾讯数据湖介绍

从上图可以看出来，整个平台比较大，包括了数据接入、上层的分析、中间的管理 (如任务管理，分析管理和引擎管理)，再到最下层的 Table Format。

二、百亿级数据落地场景落地

1. 传统平台架构

如上图所示，过去的传统平台架构无非是两种，一种是 Lambda 架构，一种是 Kappa 架构：

• Lambda 架构中，批和流是分开的，所以运维要有两套集群，一套是 For Spark/Hive，一套是 For Flink。这存在几个问题：

• • 第一是运维的成本比较大；

  • 第二是开发成本。例如在业务方面，一会要写 Spark，一会要写 Flink 或者 SQL，总体来说，开发成本对数据分析人员不是特别友好。

• 第二个是 Kappa 架构。其实就是消息队列，到底层的传输，再到后面去做一些分析。它的特点是比较快，基于 Kafka 有一定的实时性。

这两种架构各有利弊，最大的问题是存储可能会不统一，导致数据链路割裂。目前我们平台已经接入了 Iceberg，下面会根据不同场景，阐述遇到的问题及解决的过程。

2. 场景一: 手 Q 安全数据入湖

手机 QQ 安全数据入湖是一个非常典型的场景。

目前的业务场景是消息队列 TubeMQ 通过 Flink 落地成 ODS 到 Iceberg，然后再用 Flink 做一些用户表的关联，之后做成一个宽表去做一些查询，放到 COS 中，可能会在 BI 场景做一些分析。

这个过程看似平平无奇，但是要知道，手 Q 的用户关联维表为 28 亿，每天的消息队列是百亿级的，因此会面临一定的挑战。

■ 小文件挑战

1、Flink Writer 产生小文件

Flink 写入没有 shuffle，分发的数据无序，导致小文件多。

2、延迟要求高

checkpoint 间隔短，commit 间隔小，放大小文件问题。

3、小文件爆炸

几天时间元数据和数据的小文件同时爆炸，集群压力巨大。

4、合并小文件又放大问题

为了解决小文件问题，开 Action 进行小文件合并，结果产生更多文件。

5、来不及删数据

删除快照，删孤儿文件，但是扫描文件太多，namenode 压力巨大。

■  解决方案

1、Flink 同步合并

• 增加小文件合并 Operators；

• 增加 Snapshot 自动清理机制。

1）snapshot.retain-last.nums

2）snapshot.retain-last.minutes

2、Spark 异步合并

• 增加后台服务进行小文件合并和孤儿文件删除；

• 增加小文件过滤逻辑，逐步删除小文件；

• 增加按分区合并逻辑，避免一次生成太多删除文件导致任务 OOM。

■ Flink 同步合并

把所有的 Data 文件 Commit 之后，会产生一个 Commit Result。我们会拿 Commit Result 生成一个压缩的任务，再给它并发成多个 Task Manager 去做 Rewrite 的工作，最终把结果 Commit 到 Iceberg 表里面。

当然，这里面的关键所在是 CompactTaskGenerator 怎么做。刚开始的时候我们想尽量地合并，于是去做表的 scan，把很多文件都扫一遍。然而它的表非常大，小文件非常多，一扫使得整个 Flink 立马挂掉。

我们想了个方法，每次合并完，增量地去扫数据。从上一个 Replace Operation 里面到现在做一个增量，看这中间又增了多少，哪些符合 Rewrite 的策略。

这里面其实有许多配置，去看达到了多少个 snapshot，或者达到了多少个文件可以去做合并，这些地方用户可以自己设置。当然，我们本身也设有默认值，从而保证用户无感知地使用这些功能。

■ Fanout Writer 的坑

在 Fanout Writer 时，如果数据量大可能会遇到多层分区。比如手 Q 的数据分省、分市；但分完之后还是很大，于是又分 bucket。此时每个 Task Manager 里可能分到很多分区，每个分区打开一个 Writer，Writer 就会非常的多，造成内存不足。

这里我们做了两件事情：

• 第一是 KeyBy 支持。根据用户设置的分区做 KeyBy 的动作，然后把相同分区的聚集在一个 Task Manager 中，这样它就不会打开那么多分区的 Writer。当然，这样的做法会带来一些性能上的损失。

• 第二是做 LRU Writer，在内存里面维持一个 Map。

3. 场景二：新闻平台索引分析

上方是基于 Iceberg 流批一体的新闻文章在线索引架构。左边是 Spark 采集 HDFS 上面的维表，右边是接入系统，采集以后会用 Flink 和维表做一个基于 Window 的 Join，然后写到索引流水表中。

■ 功能

• 准实时明细层；

• 实时流式消费；

• 流式 MERGE INTO；

• 多维分析；

• 离线分析。

■ 场景特点

上述场景有以下几个特点：

• 数量级：索引单表超千亿，单 batch 2000 万，日均千亿；

• 时延需求：端到端数据可见性分钟级；

• 数据源：全量、准实时增量、消息流；

• 消费方式：流式消费、批加载、点查、行更新、多维分析。

■ 挑战：MERGE INTO

有用户提出了 Merge Into 的需求，因此我们从三个方面进行了思考：

• 功能：将每个 batch join 后的流水表 Merge into 到实时索引表，供下游使用；

• 性能：下游对索引时效性要求高，需要考虑 merge into 能追上上游的 batch 消费窗口；

• 易用性：Table API？还是 Action API？又或是 SQL API？

■ 解决方案

• 第一步

• 参考 Delta Lake 设计 JoinRowProcessor；

• 利用 Iceberg 的 WAP 机制写临时快照。

• 第二步

• 可选择跳过 Cardinality-check；

• 写入时可以选择只 hash，不排序。

• 第三步

• 支持 DataframeAPI；

• Spark 2.4 支持 SQL；

• Spark 3.0 使用社区版本。

4. 场景三：广告数据分析

■ 广告数据主要有以下几个特点：

• 数量级：日均千亿 PB 数据，单条 2K；

• 数据源：SparkStreaming 增量入湖；

• 数据特点：标签不停增加，schema 不停变换；

• 使用方式：交互式查询分析。

■ 遇到的挑战与对应的解决方案：

• 挑战一：Schema 嵌套复杂，平铺后近万列，一写就 OOM。

解决方案：默认每个 Parquet Page Size 设置为 1M，需要根据 Executor 内存进行 Page Size 设置。

• 挑战二：30 天数据基本集群撑爆。

  解决方案：提供 Action 进行生命周期管理，文档区分生命周期和数据生命周期。

• 挑战三：交互式查询。

  解决方案：

1）column projection；

2）predicate push down。

三、未来规划

对于未来的规划主要分为内核侧与平台侧。

1. 内核侧

在未来，我们希望在内核侧有以下几点规划：

■ 更多的数据接入

• 增量入湖支持；

• V2 Format 支持；

• Row Identity 支持。

■ 更快的查询

• 索引支持；

• Alloxio 加速层支持；

• MOR 优化。

■ 更好的数据治理

• 数据治理 Action；

• SQL Extension 支持；

• 更好的元数据管理。

2、平台侧

在平台侧我们有以下几点规划：

■ 数据治理服务化

• 元数据清理服务化；

• 数据治理服务化。

■ 增量入湖支持

• Spark 消费 CDC 入湖；

• Flink 消费 CDC 入湖。

■ 指标监控告警

• 写入数据指标；

• 小文件监控和告警。

四、总结

经过大量生产上的应用与实践，我们得到三方面的总结：

• 可用性：通过多个业务线的实战，确认 Iceberg 经得起日均百亿，甚至千亿的考验。

• 易用性：使用门槛比较高，需要做更多的工作才能让用户使用起来。
  

• 场景支持：目前支持的入湖场景 还没有 Hudi 多，增量读取这块也比较缺失，需要大家努力补齐。