图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › 技术应用› Flink CDC 原理、实践和优化
分享

Flink CDC 原理、实践和优化

Mirinda 2021-4-1 10:45:48 发表于 技术应用 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 1 3558
About云VIP帮助找工作和提升
问题导读:

1.CDC 是什么？
2.CDC 的实现原理是什么？
3.为什么选 Flink？

CDC 变更数据捕获技术可以将源数据库的增量变动记录，同步到一个或多个数据目的。本文基于腾讯云 Oceanus 提供的 Flink CDC 引擎，着重介绍 Flink 在变更数据捕获技术中的应用。
一、CDC 是什么？
CDC 是变更数据捕获（Change Data Capture）技术的缩写，它可以将源数据库（Source）的增量变动记录，同步到一个或多个数据目的（Sink）。在同步过程中，还可以对数据进行一定的处理，例如分组（GROUP BY）、多表的关联（JOIN）等。例如对于电商平台，用户的订单会实时写入到某个源数据库；A 部门需要将每分钟的实时数据简单聚合处理后保存到 Redis 中以供查询，B 部门需要将当天的数据暂存到 Elasticsearch 一份来做报表展示，C 部门也需要一份数据到 ClickHouse 做实时数仓。随着时间的推移，后续 D 部门、E 部门也会有数据分析的需求，这种场景下，传统的拷贝分发多个副本方法很不灵活，而 CDC 可以实现一份变动记录，实时处理并投递到多个目的地。下图是一个示例，通过腾讯云 Oceanus 提供的 Flink CDC 引擎，可以将某个 MySQL 的数据库表的变动记录，实时同步到下游的 Redis、Elasticsearch、ClickHouse 等多个接收端。这样大家可以各自分析自己的数据集，互不影响，同时又和上游数据保持实时的同步。
a1.png


二、CDC 的实现原理
通常来讲，CDC 分为主动查询和事件接收两种技术实现模式。对于主动查询而言，用户通常会在数据源表的某个字段中，保存上次更新的时间戳或版本号等信息，然后下游通过不断的查询和与上次的记录做对比，来确定数据是否有变动，是否需要同步。这种方式优点是不涉及数据库底层特性，实现比较通用；缺点是要对业务表做改造，且实时性不高，不能确保跟踪到所有的变更记录，且持续的频繁查询对数据库的压力较大。事件接收模式可以通过触发器（Trigger）或者日志（例如 Transaction log、Binary log、Write-ahead log 等）来实现。当数据源表发生变动时，会通过附加在表上的触发器或者 binlog 等途径，将操作记录下来。下游可以通过数据库底层的协议，订阅并消费这些事件，然后对数据库变动记录做重放，从而实现同步。这种方式的优点是实时性高，可以精确捕捉上游的各种变动；缺点是部署数据库的事件接收和解析器（例如 Debezium、Canal 等），有一定的学习和运维成本，对一些冷门的数据库支持不够。综合来看，事件接收模式整体在实时性、吞吐量方面占优，如果数据源是 MySQL、PostgreSQL、MongoDB 等常见的数据库实现，建议使用 Debezium（https://debezium.io/documentation/reference/1.4/connectors/index.html）来实现变更数据的捕获（下图来自 Debezium 官方文档   [https://debezium.io/documentation/reference/architecture.html]）。如果使用的只有 MySQL，则还可以用 Canal (https://github.com/alibaba/canal)。


a2.png

三、为什么选 Flink？
从上图可以看到，Debezium 官方架构图中，是通过 Kafka Streams 直接实现的 CDC 功能。而我们这里更建议使用 Flink CDC 模块，因为 Flink 相对 Kafka Streams 而言，有如下优势：
  • Flink 的算子和 SQL 模块更为成熟和易用
  • Flink 作业可以通过调整算子并行度的方式，轻松扩展处理能力
  • Flink 支持高级的状态后端（State Backends），允许存取海量的状态数据
  • Flink 提供更多的 Source 和 Sink 等生态支持
  • Flink 有更大的用户基数和活跃的支持社群，问题更容易解决
  • Flink 的开源协议允许云厂商进行全托管的深度定制，而 Kafka Streams 只能自行部署和运维

而且 Flink Table / SQL 模块将数据库表和变动记录流（例如 CDC 的数据流）看做是同一事物的两面(https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/table/streaming/dynamic_tables.html)，因此内部提供的 Upsert 消息结构（+I 表示新增、-U 表示记录更新前的值、+U 表示记录更新后的值，-D 表示删除）可以与 Debezium 等生成的变动记录一一对应。

四、Flink CDC 的使用方法
目前 Flink CDC 支持两种数据源输入方式。
（一）输入 Debezium 等数据流进行同步例如 MySQL -> Debezium -> Kafka -> Flink -> PostgreSQL。适用于已经部署好了 Debezium，希望暂存一部分数据到 Kafka 中以供多次消费，只需要 Flink 解析并分发到下游的场景。

a3.png

在该场景下，由于 CDC 变更记录会暂存到 Kafka 一段时间，因此可以在这期间任意启动/重启 Flink 作业进行消费；也可以部署多个 Flink 作业对这些数据同时处理并写到不同的数据目的（Sink）库表中，实现了 Source 变动与 Sink 的解耦。用法示例例如我们有个 MySQL 数据库，需要实时将内容同步到 PostgreSQL 中。假设已经安装部署好 Debezium 并开始消费 PostgreSQL 的变更日志，这些日志在持续写入名为 YourDebeziumTopic 的 Kafka 主题中。我们可以新建一个 Flink SQL 作业，然后输入如下 SQL 代码（连接参数都是虚拟的，仅供参考）：
  1. CREATE TABLE `Data_Input` (

  2.     id BIGINT,

  3.     actor VARCHAR,

  4.     alias VARCHAR,

  5.     PRIMARY KEY (`id`) NOT ENFORCED

  6. ) WITH (

  7.     'connector' = 'kafka',                             -- 可选 'kafka','kafka-0.11'. 注意选择对应的内置  Connector

  8.     'topic' = 'YourDebeziumTopic',                     -- 替换为您要消费的 Topic

  9.     'scan.startup.mode' = 'earliest-offset'            -- 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets 的任何一种

  10.     'properties.bootstrap.servers' = '10.0.1.2:9092',  -- 替换为您的 Kafka 连接地址

  11.     'properties.group.id' = 'YourGroup',               -- 必选参数, 一定要指定 Group ID

  12. 

  13.     -- 定义数据格式 (Debezium JSON 格式)

  14.     'format' = 'debezium-json',

  15.     'debezium-json.schema-include' = 'false',

  16. );

  17. 

  18. CREATE TABLE `Data_Output` (

  19.     id BIGINT,

  20.     actor VARCHAR,

  21.     alias VARCHAR,

  22.     PRIMARY KEY (`id`) NOT ENFORCED

  23. ) WITH (

  24.     'connector' = 'jdbc',

  25.     'url' = 'jdbc:postgresql://postgresql.example:50060/myDatabase?currentSchema=mySchema&reWriteBatchedInserts=true', -- 请替换为您的实际 PostgreSQL 连接参数

  26.     'table-name' = 'MyTable',   -- 需要写入的数据表

  27.     'username' = 'user',        -- 数据库访问的用户名（需要提供 INSERT 权限）

  28.     'password' = 'helloworld'   -- 数据库访问的密码

  29. );

  30. 

  31. INSERT INTO `Data_Output` SELECT * FROM `Data_Input`;
复制代码

如果在流计算 Oceanus 界面上，可以勾选 kafka 和 jdbc 两个内置的 Connector：

a4.png


随后直接开始运行作业，Flink 就会源源不断的消费 YourDebeziumTopic 这个 Kafka 主题中 Debezium 写入的记录，然后输出到下游的 MySQL 数据库中，实现了数据同步。

（二）直接对接上游数据库进行同步我们还可以跳过 Debezium 和 Kafka 的中转，使用 Flink CDC Connectors（https://github.com/ververica/flink-cdc-connectors）对上游数据源的变动进行直接的订阅处理。从内部实现上讲，Flink CDC Connectors 内置了一套 Debezium 和 Kafka 组件，但这个细节对用户屏蔽，因此用户看到的数据链路如下图所示：


a5.png


用法示例同样的，这次我们有个 MySQL 数据库，需要实时将内容同步到 PostgreSQL 中。但我们没有也不想安装 Debezium 等额外组件，那我们可以新建一个 Flink SQL 作业，然后输入如下 SQL 代码（连接参数都是虚拟的，仅供参考）：
  1. CREATE TABLE `Data_Input` (

  2.     id BIGINT,

  3.     actor VARCHAR,

  4.     alias VARCHAR,

  5.     PRIMARY KEY (`id`) NOT ENFORCED

  6. ) WITH (

  7.     'connector' = 'mysql-cdc',    -- 可选 'mysql-cdc' 和 'postgres-cdc'

  8.     'hostname' = '192.168.10.22',   -- 数据库的 IP

  9.     'port' = '3306',                -- 数据库的访问端口

  10.     'username' = 'debezium',        -- 数据库访问的用户名（需要提供 SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, SELECT, RELOAD 权限）

  11.     'password' = 'hello@world!',    -- 数据库访问的密码

  12.     'database-name' = 'YourData',   -- 需要同步的数据库

  13.     'table-name' = 'YourTable'      -- 需要同步的数据表名

  14. );

  15. 

  16. CREATE TABLE `Data_Output` (

  17.     id BIGINT,

  18.     actor VARCHAR,

  19.     alias VARCHAR,

  20.     PRIMARY KEY (`id`) NOT ENFORCED

  21. ) WITH (

  22.     'connector' = 'jdbc',

  23.     'url' = 'jdbc:postgresql://postgresql.example:50060/myDatabase?currentSchema=mySchema&reWriteBatchedInserts=true', -- 请替换为您的实际 PostgreSQL 连接参数

  24.     'table-name' = 'MyTable',   -- 需要写入的数据表

  25.     'username' = 'user',        -- 数据库访问的用户名（需要提供 INSERT 权限）

  26.     'password' = 'helloworld'   -- 数据库访问的密码

  27. );

  28. 

  29. INSERT INTO `Data_Output` SELECT * FROM `Data_Input`;
复制代码

如果在流计算页面，可以选择内置的 mysql-cdc 和 jdbc Connector：

a6.png

注意需要使用 Flink CDC Connectors(https://github.com/ververica/flink-cdc-connectors)附加组件。腾讯云 Oceanus 已经自带了 MySQL-CDC Connector，如果自行部署的话，需要下载 jar 包并将其放入 Flink 的 lib 目录下。访问数据库时，请确保连接的用户足够权限（PostgreSQL 用户看这里[https://debezium.io/documentation/reference/connectors/postgresql.html#postgresql-permissions]，MySQL 用户看这里[https://debezium.io/documentatio ... ml#setting-up-mysql]）。

五、Flink CDC 模块的实现
（一）Debezium JSON 格式解析类探秘flink-json 模块中的 org.apache.flink.formats.json.debezium.DebeziumJsonFormatFactory 是负责构造解析 Debezium JSON 格式的工厂类；同样地，org.apache.flink.formats.json.canal.CanalJsonFormatFactory 负责 Canal JSON 格式。这些类已经内置在 Flink 1.11 的发行版中，直接可以使用，无需附加任何程序包。对于 Debezium JSON 格式而言，Flink 将具体的解析逻辑放在了 org.apache.flink.formats.json.debezium.DebeziumJsonDeserializationSchema#DebeziumJsonDeserializationSchema 类中。


a7.png

上图表示 Debezium JSON 的一条更新（Update）消息，它表示上游已将 id=123 的数据更新，且字段内包含了更新前的旧值，以及更新后的新值。那么，Flink 是如何解析并生成对应的 Flink 消息呢？我们看下这个类的 deserialize 方法：
  1. GenericRowData before = (GenericRowData) payload.getField(0);  // 更新前的数据
复制代码

从上述逻辑可以看出，对于每一种 Debezium 的操作码（op 字段的类型），都可以用 Flink 的 RowKind 类型来表示。对于插入 +I 和删除 D，都只需要一条消息即可；而对于更新，则涉及删除旧数据和写入新数据，因此需要 -U 和 +U 两条消息来对应。特别地，在 MySQL、PostgreSQL 等支持 Upsert（原子操作的 Update or Insert）语义的数据库中，通常前一个 -U 消息可以省略，只把后一个 +U 消息用作实际的更新操作即可，这个优化在 Flink 中也有实现。因此可以看到，Debezium 到 Flink 消息的转换逻辑是非常简单和自然的，这也多亏了 Flink 先进的设计理念，很早就提出并实现了 Upsert 数据流和动态数据表之间的映射关系。

1.Flink CDC Connectors 的实现
（1）flink-connector-debezium 模块
我们在使用 Flink CDC Connectors 时，也会好奇它究竟是如何做到的不需要安装和部署外部服务就可以实现 CDC 的。当我们阅读  flink-connector-mysql-cdc 的源码时，可以看到它内部依赖了 flink-connector-debezium 模块，而这个模块将 Debezium Embedded（https://github.com/debezium/debezium/tree/master/debezium-embedded）嵌入到了 Connector 中。flink-connector-debezium 的数据源实现类为 com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction，它集成了 Flink 中的 RichSourceFunction 并实现了 CheckpointedFunction 以支持快照保存状态。通常而言，对于 SourceFunction，我们可以从它的 run 方法入手分析。它的核心代码如下：
  1. this.engine = DebeziumEngine.create(Connect.class)    .using(properties) // 初始化 Debezium 所需的参数 
复制代码

可以看到，这个 SourceFunction 使用一些预先定义的参数，初始化了一个嵌入式的 DebeziumEngine（Java 的 Runnable），然后提交给线程池（executor）去执行。这个 Debezium 线程会批量接收 binlog 信息并回调传入的 debeziumConsumer 以反序列化消息并交给 Flink 来处理。本类的其他方法主要负责初始化状态和保存快照，这里略过。这里我们再来看一下 DebeziumChangeConsumer 的实现，它的最核心的方法是 handleBatch 。当 Debezium 收到一批新的事件时，会调用这个方法来通知我们的 Connector 进行处理。这里有个 for 循环轮询的逻辑：
  1. for (ChangeEvent<SourceRecord, SourceRecord> event : changeEvents) {  // 轮询各个事件

  2.     SourceRecord record = event.value();

  3.     if (isHeartbeatEvent(record)) {    // 如果时心跳包

  4.         // 只更新当前 offset 信息, 然后继续（不进行实际处理）

  5.         synchronized (checkpointLock) {

  6.             debeziumOffset.setSourcePartition(record.sourcePartition());

  7.             debeziumOffset.setSourceOffset(record.sourceOffset());

  8.         }

  9.         continue;

  10.     }

  11. 

  12.     deserialization.deserialize(record, debeziumCollector);    // 反序列化这条消息

  13. 

  14.     if (isInDbSnapshotPhase) {  // 如果处于数据库快照期, 需要阻止 Flink 检查点（Checkpoint）生成

  15.         if (!lockHold) {

  16.             MemoryUtils.UNSAFE.monitorEnter(checkpointLock);

  17.             lockHold = true;

  18.             ...

  19.         }

  20.         if (!isSnapshotRecord(record)) {  // 如果已经不在数据库快照期了, 就释放锁, 允许 Flink 正常生成检查点（Checkpoint）

  21.             MemoryUtils.UNSAFE.monitorExit(checkpointLock);

  22.             isInDbSnapshotPhase = false;

  23.             ...

  24.         }

  25.     }

  26. 

  27.   // 更新当前 offset 信息, 并向下游 Flink 算子发送数据

  28.     emitRecordsUnderCheckpointLock(

  29.         debeziumCollector.records, record.sourcePartition(), record.sourceOffset());

  30. }

复制代码

可以看到逻辑比较简单，只需要关注 checkpointLock 这个对象：只有持有这个对象的锁时，才允许 Flink 进行检查点的生成。当作业处于数据库快照期（即作业刚启动时，需全量同步源数据库的一份完整快照，此时收到的数据类型是 Debezium 的 SnapshotRecord），则不允许 Flink 进行 Checkpoint 即检查点的生成，以避免作业崩溃恢复后状态不一致；同样地，如果正在向下游算子发送数据并更新 offset 信息时，也不允许快照的进行。这些操作都是为了保证 Exacly-Once（精确一致）语义。这里也解释了在作业刚启动时，如果数据库较大（同步时间较久），Flink 刚开始的 Checkpoint 永远失败（超时）的原因：只有当 Flink 完整同步了全量数据后，才可以进行增量数据的处理，以及 Checkpoint 的生成。

（2）flink-connector-mysql-cdc 模块
而对于 flink-connector-mysql-cdc 模块而言，它主要涉及到 MySQLTableSource 的声明和实现。我们知道，Flink 是通过 Java 的 SPI（Service Provider Interface）机制动态加载 Connector 的，因此我们首先看这个模块的 src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory 文件，里面内容指向 com.alibaba.ververica.cdc.connectors.mysql.table.MySQLTableSourceFactory。打开这个工厂类，我们可以看到它定义了该 Connector 所需的参数，例如 MySQL 数据库的用户名、密码、表名等信息，并负责 MySQLTableSource 实例的具体创建，而 MySQLTableSource 类对这些参数做转换，最终会生成一个上文提到的 DebeziumSourceFunction 对象。因此我们可以发现，这个模块作用是一个 MySQL 参数的封装和转换层，最终的逻辑实现仍然是由 flink-connector-debezium 完成的。

六、MySQL CDC 常见问题&优化
由于 Flink 的 CDC 功能还比较新（1.11 版本刚开始支持，1.12 版本逐步完善），因而在应用过程中，很可能会遇到有各种问题。鉴于大多数客户的数据源都是 MySQL，我们这里整理了客户常见的一些问题和优化方案，希望能够帮助到大家。Debezium 报错：binlog probably contains events generated with statement or mixed based replication format当前的 Binlog 格式被设置为了 STATEMENT 或者 MIXED， 这两种都不被 Debezium 支持。为了使用 Flink CDC 功能，需要把 MySQL 的 binlog-format 设置为 ROW：
  1. SET GLOBAL binlog_format = 'ROW';
复制代码

如果您使用的是腾讯云的 TencentDB for MySQL，请确认下面设置：


a8.png


Debezium 报错：User does not have the 'LOCK TABLES' privilege required to obtain a consistent snapshot 或 Access denied; you need (at least one of) the SUPER, REPLICATION CLIENT privilege(s)
请对作业中指定的 MySQL 用户赋予如下权限：SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT，例如：
  1. GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '用户名' IDENTIFIED BY '密码';

  2. FLUSH PRIVILEGES;
复制代码

如果您使用的数据库不允许或者不希望使用 RELOAD 进行全局锁，则还需要授予 LOCK TABLES 权限以令 Debezium 尝试进行表级锁。注意，表级锁会导致更长的数据库锁定时间！

如果希望彻底跳过锁（对数据的一致性要求不高，但要求数据库不能被锁），则可以在 WITH 参数中设置 'debezium.snapshot.locking.mode' = 'none' 参数来跳过锁操作。但请注意，同步过程中千万不要随意变更库表的结构。作业刚启动期间，Flink Checkpoint 一直失败/重启前文讲过，Flink CDC Connector 在初始的全量快照同步阶段，会屏蔽掉快照的执行，因此如果 Flink Checkpoint 需要执行的话，就会因为一直无法获得 checkpointLock 对象的锁而超时。可以设置 Flink 的 execution.checkpointing.tolerable-failed-checkpoint 参数以容忍更多的 Checkpoint 失败事件，同时可以调大 Checkpoint 周期，避免作业因 Checkpoint 失败而一直重启。JDBC Sink 批量写入时，数据会缺失几条如果发现数据库中的某些数据在 CDC 同步后有缺失，请确认是否仍在使用 Flink 旧版 1.10 的 Flink SQL WITH 语法（例如 WITH 参数中的 connector.type 是旧语法[https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/connect.html#jdbc-connector]，connector 是新语法[https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/connectors/jdbc.html#how-to-create-a-jdbc-table]）。旧版语法的 Connector 在 JDBC 批量写入 Upsert 数据（例如数据库的更新记录）时，并未考虑到 Upsert 与 Delete 消息之间的顺序关系，因此会出现错乱的问题，请尽快迁移到新版的 Flink SQL 语法。异常数据造成作业持续重启默认情况下，如果遇到异常的数据（例如消费的 Kafka topic 在无意间混入了其他数据），Flink 会立刻崩溃重启，然后从上个快照点（Checkpoint）重新消费。由于某条异常数据的存在，作业会永远因为异常而重启。可以在 WITH 参数中加入 'debezium-json.ignore-parse-errors' = 'true' 来应对这个问题。上游 Debezium 崩溃导致写入重复数据，结果不准Debezium 服务端发生异常并恢复后，由于可能没有及时记录崩溃前的现场，可能会退化为 At least once 模式，即同样的数据可能被发送多次，造成下游结果不准确。为了应对这个问题，新版的 Flink 1.12 增加了一个 table.exec.source.cdc-events-duplicate 配置项（可以编辑 flink-conf.yaml 文件来配置），建议将其设置为 true 以对这些重复数据进行去重。但是需要注意，该选项需要数据源表定义了主键，否则也无法进行去重操作。

七、未来展望
在 Flink 1.11 版本中，CDC 功能首次被集成到内核中。由于 Flink 1.11.0 版本有个 严重 Bug（https://issues.apache.org/jira/browse/FLINK-18461）造成 Upsert 数据无法写入下游，我们建议使用 1.11.1 及以上版本。在 1.12 版本上，Flink 还在配置项中增加了前文提到的 table.exec.source.cdc-events-duplicate 等选项以更好地支持 CDC 去重；还支持 Avro 格式的 Debezium 数据流，而不仅仅限于 JSON 了。另外，这个版本增加了对 Maxwell(https://maxwells-daemon.io/)格式的 CDC 数据流支持，为了更好地完善 CDC 功能模块，Flink 社区创建了 [FLINK-18822] 以追踪关于该模块的进展。可以从中看到，Flink 1.13 主要着力于支持更多的类型（FLINK-18758[https://issues.apache.org/jira/browse/FLINK-18758]），以及允许从 Debezium Avro、Canal 等数据流中读取一些元数据信息等。而在更远的规划中，Flink 还可能支持基于 CDC 的内存数据库缓存，这样我们可以在内存中动态地 JOIN 一个数据库的副本，而不必每次都查询源库，这将极大地提升作业的处理能力，并降低数据库的查询压力。

最新经典文章，欢迎关注公众号


原文链接: https://mp.weixin.qq.com/s/_khnjLbX56Ud4yYFvHnmkA


回复

使用道具 举报

自定义广告语

已有(1)人评论

跳转到指定楼层
正序浏览
若无梦何远方 发表于 2021-4-4 14:42:05
Flink CDC
&#8226; 是变更数据获取的缩写 （Change Data Capture）
&#8226; 将源数据 （Source）的变动同步到一个或者多个目的地 (Sink)
&#8226; 再同步的过程中可以对数据进行一些额外操作 （GROUP BY | JOIN | ...）
&#8226; 解决: 对同一份数据多次下沉的场景，解决传统方式对一份数据多次分发的操作
&#8226; 比如落地： Redis | Elasticsearch | ClickHouse
实现模式
主动查询
&#8226; 在数据源表中 保存上次更新的时间戳跟版本号
&#8226; 下游通过不断查询和上次记录做对比，来确定数据是否有变动，是否需要同步
&#8226; 优点：不涉及底层 binlog 的采集与解析 （减少数据传输过程）
&#8226; 缺点：下游需要不断查询对数据库 Service 有压力
事件接收
&#8226; 通过触发器 Trigger 或者 日志的操作记录该事件 （例如 Transaction log、Binary log、Write-ahead log）
&#8226; 源数据发生改变后附加在 Trigger 或者 LOG(Binlog) 中
&#8226; 下游通过数据库底层的协议订阅并消费这些事件，对数据库的变动做回放的效果 达到数据的同步工作
&#8226; 优点：可以减少对数据库 Service 的压力，实时性比较高
&#8226; 缺点：针对于 LOG 需要有对应的解析工具参与解析 （Canal、Datax、Debezium）,学习成本高
使用方法
输入 Debezium 等数据流进行同步
&#8226; 特点：做到处理的解耦工作 | 对于数据的处理得到的解耦合
相当于是 DBServer ->  Debezium -> Kafka -> Flink -> TransSink
&#8226; 缺点：个人感觉维护量大、学习成本高
直接对接上游数据库进行数据同步
&#8226; 特点：学习成本小、无需考虑内部细节（内部都做了封装、开箱即用）
相当于是 DBServer -> Flink CDC ->TransSink
&#8226; 缺点：不透明化、内聚、数据安全需要考虑 （类似做一些加密操作）
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
Mirinda

超级版主

关注

  • 123

    主题

  • 125

    帖子

  • 11

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000