导读：本文基于 Flink + Apache Doris 构建实时数仓的业务场景的调研后果，根据用户的面临的挑战和问题对 Flink 实时写入 Apache Doris 的优化实现与将来布局进行了具体的介绍。
背景
随着数据实时化需要的日益增多，数据的时效性对企业的精细化经营越来越重要，在海量数据中，如何能实时无效的挖掘出有价值的信息，疾速的获取数据反馈，帮助公司更快的做出决策，更好的进行产品迭代，实时数仓在这一过程中起到了不可代替的作用。
在这种局势下，Apache Doris 作为一款实时 MPP 剖析型数据库怀才不遇，同时具备高性能、简略易用等个性，具备丰盛的数据接入形式，联合 Flink 流式计算，能够让用户疾速将 Kafka 中的非结构化数据以及 MySQL 等上游业务库中的变更数据，疾速同步到 Doris 实时数仓中，同时 Doris 提供亚秒级剖析查问的能力，能够无效地满足实时 OLAP、实时数据看板以及实时数据服务等场景的需要。
挑战
通常实时数仓要保障端到端高并发以及低提早，往往面临诸多挑战，比方：
- 如何保障端到端的 秒级别数据同步？
- 如何疾速保障 数据可见性？
- 在高并发大压力下，如何解决 大量小文件写入 的问题？
- 如何确保端到端的 Exactly Once 语义？
联合这些挑战，同时对用户应用 Flink+Doris 构建实时数仓的业务场景进行深刻调研，在把握了用户应用的痛点之后，咱们在 Doris 1.1 版本中进行了针对性的优化，大幅晋升实时数仓构建的用户体验，同时晋升零碎的稳定性，系统资源耗费也失去了大幅的优化。
优化
流式写入
Flink Doris Connector 最后的做法是在接管到数据后，缓存到内存 Batch 中，通过攒批的形式进行写入，同时应用 batch.size、batch.interval 等参数来管制 Stream Load 写入的机会。这种形式通常在参数正当的状况下能够稳固运行，一旦参数不合理导致频繁的 Stream Load，便会引发 Compaction 不及时，从而导致 version 过多的谬误 (-235)；其次，当数据过多时，为了缩小 Stream Load 的写入机会，batch.size 过大的设置还可能会引发 Flink 工作的 OOM。为了解决这个问题， 咱们引入了流式写入 ：
- Flink 工作启动后，会异步发动一个 Stream Load 的 Http 申请。
- 接管到实时数据后，通过 Http 的分块传输编码 (Chunked transfer encoding) 机制继续向 Doris 传输数据。3. 在 Checkpoint 时完结 Http 申请，实现本次 Stream Load 写入，同时异步发动下一次 Stream Load 的申请。4. 持续接管实时数据，后续流程同上。
因为采纳 Chunked 机制传输数据，就防止了攒批对内存的压力，同时将写入的机会和 Checkpoint 绑定起来，使得 Stream Load 的机会可控，并且为上面的 Exactly-Once 语义提供了根底。
Exactly-Once
Exactly-Once 语义是指即便在机器或利用呈现故障的状况下，也不会反复解决数据或者失落数据。Flink 很早就反对 End-to-End 的 Exactly-Once 场景，次要是通过两阶段提交协定来实现 Sink 算子的 Exactly-Once 语义。在 Flink 两阶段提交的根底上，同时借助 Doris 1.0 的 Stream Load 两阶段提交，Flink Doris Connector 实现了 Exactly Once 语义，具体原理如下：
- Flink 工作在启动的时候，会发动一个 Stream Load 的 PreCommit 申请，此时会先开启一个事务，同时会通过 Http 的 Chunked 机制将数据继续发送到 Doris。
- 在 Checkpoint 时，完结数据写入，同时实现 Http 申请，并且将事务状态设置为预提交(PreCommitted)，此时数据曾经写入 BE，对用户不可见。
- Checkpoint 实现后，发动 Commit 申请，并且将事务状态设置为提交(Committed)，实现后数据对用户可见。
- Flink 利用意外挂掉后，从 Checkpoint 重启时，若上次事务为预提交 (PreCommitted) 状态，则会发动回滚申请，并且将事务状态设置为 Aborted。
基于此，能够借助 Flink Doris Connector 实现数据实时入库时数据不丢不重。
秒级别数据同步
高并发写入场景下的端到端秒级别数据同步以及数据的实时可见能力，须要 Doris 具备如下几方面的能力：
事务处理能力
Flink 实时写入以 Stream Load 2PC 的形式与 Doris 进行交互，须要 Doris 具备对应的事务处理能力，保障事务根本的 ACID 个性，在高并发场景下撑持 Flink 秒级别的数据同步。
数据版本的疾速聚合能力
Doris 外面一次导入会产生一个数据版本，在高并发写入场景下必然带来的一个影响是数据版本过多，且单次导入的数据量不会太大。继续的高并发小文件写入场景对 Doris 并不敌对，极其考验 Doris 数据合并的实时性以及性能，进而会影响到查问的性能。Doris 在 1.1 中大幅加强了数据 Compaction 能力，对于新增数据可能疾速实现聚合，防止分片数据中的版本过多导致的 -235 谬误以及带来的查问效率问题。 首先 ，在 Doris 1.1 版本中，引入了 QuickCompaction，减少了被动触发式的 Compaction 查看，在数据版本减少的时候被动触发 Compaction。同时通过晋升分片元信息扫描的能力，疾速的发现数据版本多的分片，触发 Compaction。通过主动式触发加被动式扫描的形式，彻底解决数据合并的实时性问题。 同时 ，针对高频的小文件 Cumulative Compaction，实现了 Compaction 工作的调度隔离，避免重量级的 Base Compaction 对新增数据的合并造成影响。 最初，针对小文件合并，优化了小文件合并的策略，采纳梯度合并的形式，每次参加合并的文件都属于同一个数据量级，避免大小差异很大的版本进行合并，逐步有档次的合并，缩小单个文件参加合并的次数，可能大幅的节俭零碎的 CPU 耗费。
Doris 1.1 对高并发导入、秒级别数据同步、数据实时可见等场景都做了针对性优化，大大增加了 Flink + Doris 零碎的易用性以及稳定性，节俭了集群整体资源。
成果
通用 Flink 高并发场景
在调研的通用场景中，应用 Flink 同步上游 Kafka 中的非结构化数据，通过 ETL 后应用 Flink Doris Connector 将数据实时写入 Doris 中。这里客户场景极其严苛，上游维持以每秒 10w 的超高频率写入，须要数据可能在 5s 内实现上下游同步，实现秒级别的数据可见。这里 Flink 配置为 20 并发，Checkpoint 距离 5s，Doris 1.1 的体现相当优异。具体体现在如下几个方面：
Compaction 实时性
数据能疾速合并，Tablet 数据版本个数维持在 50 以下，Compaction Score 稳固。相比于之前高并发导入频出的 -235 问题，Compaction 合并效率有 10+ 倍晋升。
CPU 资源耗费
Doris 1.1 针对小文件的 Compaction 进行了策略优化，在上述高并发导入场景，CPU 资源耗费降落 25%。
QPS 查问提早稳固
通过升高 CPU 使用率，缩小数据版本的个数，晋升了数据整体有序性，从而缩小了 SQL 查问的提早。
秒级别数据同步场景（极限大压力）
单 BE 单 Tablet，客户端 30 并发极限 Stream Load 压测，数据在实时性 <1s，Compaction Score 优化前后比照
应用倡议
数据实时可见场景
对提早要求特地严格的场景，比方秒级别数据同步，通常意味着单次导入文件较小，此时倡议调小 cumulative_size_based_promotion_min_size_mbytes，单位是 MB，默认 64，能够设置成 8，可能很大水平晋升 Compaction 的实时性。
高并发场景
对于高并发的写入场景，能够通过减少 Checkpoint 的距离来缩小 Stream Load 的频率，比方 Checkpoint 能够设置为 5-10s，不仅能够减少 Flink 工作的吞吐，也能够缩小小文件的产生，防止给 Compaction 造成更多压力。
此外，对数据实时性要求不高的场景，比方分钟级别的数据同步，能够减少 Checkpoint 的距离，比方 5-10 分钟，此时 Flink Doris Connector 仍然可能通过两阶段提交 +checkpoint 机制来保证数据的完整性。
将来布局
实时 Schema Change
目前通过 Flink CDC 实时接入数据时，当上游业务表进行 Schema Change 操作时，必须先手动批改 Doris 中的 Schema 和 Flink 工作中的 Schema，最初再重启工作，新的 Schema 的数据才能够同步过去。这样应用形式须要人为的染指，会给用户带来极大的运维累赘。后续会针对 CDC 场景做到反对 Schema 实时变更，上游的 Schema Change 实时同步到上游，全面晋升 Schema Change 的效率。
Doris 多表写入
目前 Doris Sink 算子仅反对同步单张表，所以对于整库同步的操作，须要手动在 Flink 层面进行分流，写到多个 Doris Sink 中，这无疑减少了开发者的难度，在后续版本中咱们也将反对单个 Doris Sink 同步多张表，这样就大大的简化了用户的操作。
自适应的 Compaction 参数调优
目前 Compaction 策略参数较多，在大部分通用场景能施展较好的成果，然而在一些非凡场景下并不能高效的发挥作用。咱们将在后续版本中继续优化，针对不同的场景，进行自适应的 Compaction 调优，在各类场景下进步数据合并效率，晋升实时性。
单正本 Compaction
目前的 Compaction 策略是各 BE 独自进行，在后续版本中咱们将实现单正本 Compaction，通过克隆快照的形式实现 Compaction 工作，缩小集群 2/3 的 Compaction 工作，升高零碎的负载，把更多的系统资源留给用户侧。
退出社区
如果你对 Apache Doris 感兴趣，请点击 “ 浏览原文 ” 理解并退出 Doris！咱们也发动了征文活动 ， 邀你讲讲与 Doris“ 相遇 相知 相识 ”的故事，不仅有精美礼品相送，还可取得 SelectDB 全渠道 流传曝光加持！最初，欢送更多的开源技术爱好者退出 Apache Doris 社区，携手成长，共建社区生态。
相干链接：
SelectDB 官方网站：
https://selectdb.com
Apache Doris 官方网站：
http://doris.apache.org
Apache Doris Github：
https://github.com/apache/doris
Apache Doris 开发者邮件组：
dev@doris.apache.org