《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 的生产实践(上) /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天前
|
消息中间件 关系型数据库 MySQL
实时计算 Flink版操作报错合集之遇到报错:Apache Kafka Connect错误如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
12 5
|
1天前
|
SQL 关系型数据库 MySQL
实时计算 Flink版操作报错合集之报错:org.apache.flink.table.api.validationexception如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
8 1
|
1天前
|
存储 SQL 关系型数据库
实时计算 Flink版操作报错合集之报错:WARN (org.apache.kafka.clients.consumer.ConsumerConfig:logUnused)这个错误如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
13 3
|
2天前
|
敏捷开发 运维 监控
深入探索软件测试中的持续集成与持续部署(CI/CD)实践
【5月更文挑战第17天】 在现代软件开发过程中,"持续集成"(Continuous Integration, CI)和"持续部署"(Continuous Deployment, CD)已经成为提高开发效率、确保代码质量和加速产品上市速度的关键策略。本文将深入探讨CI/CD在软件测试中的应用,包括它们的定义、目的、实施策略以及面临的挑战。通过对自动化测试、版本控制、构建流程和反馈循环的详细分析,我们将揭示如何利用CI/CD实践来优化测试流程,减少错误,并确保高质量的软件交付。
|
2天前
|
Oracle NoSQL 关系型数据库
实时计算 Flink版产品使用合集之MongoDB CDC connector的全量快照功能可以并发读取吗
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
24 2
|
3天前
|
SQL 资源调度 NoSQL
实时计算 Flink版产品使用合集之使用Flink CDC SQL MongoDB Connector时,可以采取什么措施来提升数据消费速率
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
22 0
|
3天前
|
Java 数据库连接 Spring
K8S+Docker理论与实践深度集成java面试jvm原理
K8S+Docker理论与实践深度集成java面试jvm原理
|
4天前
|
运维 监控 Kubernetes
构建高效自动化运维体系:基于容器技术的持续集成与持续部署(CI/CD)实践
【5月更文挑战第15天】 随着云计算和微服务架构的普及,传统的IT运维模式面临转型压力。为提高软件交付效率并降低运维成本,本文探讨了利用容器技术实现自动化运维的有效策略。重点分析了在持续集成(CI)和持续部署(CD)流程中,容器如何发挥作用,以及它们如何帮助组织实现敏捷性和弹性。通过具体案例研究,文章展示了容器化技术在自动化测试、部署及扩展中的应用,并讨论了其对系统稳定性和安全性的影响。
|
4天前
|
运维 监控 安全
构建高效自动化运维系统:基于容器技术的持续集成与持续部署(CI/CD)实践
【5月更文挑战第14天】 随着DevOps文化的深入人心,持续集成与持续部署(CI/CD)已成为现代软件工程不可或缺的组成部分。本文将探讨如何利用容器技术,尤其是Docker和Kubernetes,构建一个高效、可扩展的自动化运维系统。通过深入分析CI/CD流程的关键组件,我们将讨论如何整合这些组件以实现代码从提交到生产环境的快速、无缝过渡。文章还将涉及监控、日志管理以及安全性策略等运维考量,为读者提供一个全面的自动化运维解决方案蓝图。
|
4天前
|
JSON NoSQL MongoDB
理解Nosql数据库的mongodb
【5月更文挑战第5天】MongoDB是2009年发布的一款通用型NoSQL数据库,结合了关系模型和NoSQL的优点,适用于各种现代应用。其特点包括图形界面、数据服务、云基础设施集成(AWS, Azure, Google Cloud)。它具备全面的查询能力、ACID事务、可调整的一致性保证,并有多语言驱动及工具,可在任何地方运行。
59 4

热门文章

最新文章

  • 1
    实时计算 Flink版操作报错合集之遇到报错:"An OperatorEvent from an OperatorCoordinator to a task was lost. Triggering task failover to ensure consistency." ,该怎么办
    8
  • 2
    实时计算 Flink版操作报错合集之在连接Oracle 19c时报错如何解决
    8
  • 3
    实时计算 Flink版操作报错合集之写入 Kafka 报错 "Failed to send data to Kafka: Failed to allocate memory within the configured max blocking time 60000 ms",该怎么解决
    10
  • 4
    实时计算 Flink版操作报错合集之报错显示“Unsupported SQL query! sqlUpdate() only accepts SQL statements of type INSERT and DELETE"是什么意思
    9
  • 5
    实时计算 Flink版操作报错合集之报错io.debezium.DebeziumException: The db history topic or its content is fully or partially missing. Please check database history topic configuration and re-execute the snapshot. 是什么原因
    8
  • 6
    实时计算 Flink版操作报错合集之本地打成jar包,运行报错,idea运行不报错,是什么导致的
    7
  • 7
    实时计算 Flink版操作报错合集之使用 Event Time Temporal Join 关联多个 HBase 后,Kafka 数据的某个字段变为 null 是什么原因导致的
    8
  • 8
    实时计算 Flink版操作报错合集之使用 Event Time Temporal Join 关联多个 HBase 后,Kafka 数据的某个字段变为 null 是什么原因导致的
    10
  • 9
    实时计算 Flink版操作报错合集之查询sqlserver ,全量阶段出现报错如何解决
    10
  • 10
    实时计算 Flink版操作报错合集之执行Flink job,报错“Could not execute SQL statement. Reason:org.apache.flink.table.api.ValidationException: One or more required options are missing”,该怎么办
    8
  • 相关产品

  • 实时计算 Flink版
  • http://www.vxiaotou.com