《Apache Flink 案例集(2022版)》——1.数据集成——XTransfer-基Flink MongoDB CDC 在 XTransfer 的生产实践(下)

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 《Apache Flink 案例集(2022版)》——1.数据集成——XTransfer-基Flink MongoDB CDC 在 XTransfer 的生产实践(下)

《Apache Flink 案例集(2022版)》——1.数据集成——XTransfer-基Flink MongoDB CDC 在 XTransfer 的生产实践(上) https://developer.aliyun.com/article/1228399



生产实践

在实际的生产使用过程当中,XTransfer积累了如下的实践经验:


1. 使用 RocksDB State Backend

Changelog Normalize 为了补齐 -U 的前置镜像值,会带来额外的状态开销,在生产环境中推荐使用 RocksDB State Backend。  


2. 合适的 oplog 容量和过期时间

MongoDB oplog.rs 是一个特殊的有容量集合,当 oplog.rs 容量达到最大值时,会丢弃历史的数据。Change Streams 通过 resume token 进行恢复,太小的 oplog 容量可能导致 resume token 对应的 oplog 记录不再存在,因而导致恢复失败。


在没有显示指定 oplog 容量时,WiredTiger 引擎的 oplog 默认容量为磁盘大小的 5%,下限为 990MB,上限为 50GB。在 MongoDB 4.4 之后,支持设置 oplog 最短保留时间,在 oplog 已满并且 oplog 记录超过最短保留时间时,才会对该 oplog 记录进行回收。


可以使用 replSetResizeOplog 命令重新设置 oplog 容量和最短保留时间。在生产环境下,建议设置 oplog 容量不小于 20GB,oplog 保留时间不少于 7 天。


3. 变更慢的表开启心跳事件

Flink MongoDB CDC 会定期将 resume token 写入 checkpoint 对 Change Stream 进行恢复,MongoDB 变更事件或者心跳事件都能触发 resume token 的更新。如果订阅的集合变更缓慢,可能造成最后一条变更记录对应的 resume token 过期,从而无法从 checkpoint 进行恢复。因此对于变更缓慢的集合,建议开启心跳事件 (设置 heartbeat.interval.ms > 0),来维持 resume token 的更新。  


4. 自定义 MongoDB 连接参数

当默认连接无法满足使用要求时,可以通过 connection.options 配置项传递 MongoDB 支持的连接参数。  


5. Change Stream 参数调优

可以在 Flink DDL 中通过 poll.await.time.ms 和 poll.max.batch.size 精细化配置变更事件的拉取。


poll.await.time.ms

变更事件拉取时间间隔,默认为 1500ms。对于变更频繁的集合,可以适当调小拉取间隔,提升处理时效;对于变更缓慢的集合,可以适当调大拉取时间间隔,减轻数据库压力。


poll.max.batch.size

每一批次拉取变更事件的最大条数,默认为 1000 条。调大改参数会加快从 Cursor 中拉取变更事件的速度,但会提升内存的开销。  


6. 订阅整库、集群变更

在2.2.0版本之前可以设置database = "db",collection = "",订阅 db 整库的变更;database = "",collection = "",可以订阅整个集群的变更。


在2.2.0版本及之后支持了正则过滤数据库和集合的功能,例如database = "db1|db2",collection = "db1.products|db2.orders"。  


7. 权限控制

MongoDB 支持对用户、角色、权限进行细粒度的管控,开启 Change Stream 的用户需要拥有 find 和 changeStream 两个权限;


如果对单库订阅,则需要相应数据库的 read 角色。


如果同时对库、集合使用正则过滤,则需要 readAnyDatabase 角色。


在生产环境下,建议创建 Flink 用户和角色,并对该角色进行细粒度的授权。需要注意的是,MongoDB 可以在任何 database 下创建用户和角色,如果用户不是创建在 admin 下,需要在连接参数中指定 authSource =< 用户所在的 database>;


在开发环境和测试环境下,可以授予 read 和 readAnyDatabase 两个内置角色给 Flink 用户,即可对任意集合开启 change stream。


未来规划


1. 支持增量 Snapshot

目前,MongoDB CDC Connector 还不支持增量 Snapshot,对于数据量较大的表还不能很好发挥 Flink 并行计算的优势。后续将实现 MongoDB 的增量 Snapshot 功能,使其支持 Snapshot 阶段的 checkpoint,和并发度设置。  


2. 支持从指定时间进行变更订阅

目前,MongoDB CDC Connector 仅支持从当前时间开始 Change Stream 的订阅,后续将提供从指定时间点的 Change Stream 订阅。

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
1月前
|
监控 Oracle 关系型数据库
Dataphin实时集成Oracle CDC相关问题排查
本文档提供了Dataphin平台Oracle CDC实时集成相关问题排查指南,覆盖了权限等常见问题,旨在帮助快速定位和解决Oracle数据库变更数据捕获(CDC)集成过程中所可能遇到的技术难题,确保数据的实时、准确同步。
|
1月前
|
消息中间件 API Apache
官宣|阿里巴巴捐赠的 Flink CDC 项目正式加入 Apache 基金会
本文整理自阿里云开源大数据平台徐榜江 (雪尽),关于阿里巴巴捐赠的 Flink CDC 项目正式加入 Apache 基金会。
1411 1
官宣|阿里巴巴捐赠的 Flink CDC 项目正式加入 Apache 基金会
|
1月前
|
SQL Apache 流计算
Apache Flink官方网站提供了关于如何使用Docker进行Flink CDC测试的文档
【2月更文挑战第25天】Apache Flink官方网站提供了关于如何使用Docker进行Flink CDC测试的文档
142 3
|
1月前
|
Oracle 关系型数据库 流计算
flink cdc 同步问题之报错org.apache.flink.util.SerializedThrowable:如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
1月前
|
SQL 分布式计算 NoSQL
使用Apache Hudi和Debezium构建健壮的CDC管道
使用Apache Hudi和Debezium构建健壮的CDC管道
16 0
|
1月前
|
分布式计算 Java 数据管理
使用Apache Hudi + Amazon EMR进行变化数据捕获(CDC)
使用Apache Hudi + Amazon EMR进行变化数据捕获(CDC)
87 0
|
1月前
|
缓存 分布式计算 Apache
Apache Hudi与Apache Flink更好地集成,最新方案了解下?
Apache Hudi与Apache Flink更好地集成,最新方案了解下?
60 0
|
1月前
|
监控 Apache 开发工具
Apache Flink 1.12.2集成Hudi 0.9.0运行指南
Apache Flink 1.12.2集成Hudi 0.9.0运行指南
67 0
|
1月前
|
存储 SQL 分布式计算
Apache Hudi在Linkflow构建实时数据湖的生产实践
Apache Hudi在Linkflow构建实时数据湖的生产实践
39 0
|
1月前
|
SQL API 数据处理
新一代实时数据集成框架 Flink CDC 3.0 —— 核心技术架构解析
本文整理自阿里云开源大数据平台吕宴全关于新一代实时数据集成框架 Flink CDC 3.0 的核心技术架构解析。
714 0
新一代实时数据集成框架 Flink CDC 3.0 —— 核心技术架构解析

相关产品

  • 实时计算 Flink版