Apache Flink CDC 3.1.0 发布公告

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: Apache Flink 社区很高兴地宣布发布 Flink CDC 3.1.0!

Apache Flink 社区很高兴地宣布发布 Flink CDC 3.1.0!这是社区在接受 Flink CDC 作为 Apache Flink 的子项目后的首个版本,带来了令人兴奋的新功能,如 transformation 的支持和分库分表合并。Flink CDC 的生态系统也在不断扩展,包括新增的 Pipeline 连接器 Kafka 和 Paimon 以及对已有的 Source 连接器增强。

欢迎浏览 Flink CDC 文档和尝试快速入门教程来探索 Flink CDC 的世界!同时欢迎下载该版本并通过 Flink 的邮件列表JIRA在 Flink 社区进行讨论和分享,期待大家的反馈!

Flink CDC 3.1 快速预览

作为 Flink CDC 成为 Apache Flink 子项目之后的首个版本,3.1 带来了许多新功能与稳定性提升。主要亮点包括:

  1. Transformation 支持:通过 YAML 管道定义中的 transform 部分,用户可以对数据变化事件进行投影、计算和添加常量列等转化,使用类似 SQL 的语法,提升数据集成管道的灵活性。
  2. 分库分表合并支持:可以通过在 YAML 管道定义中配置路由将多个表合并到一个目标表,自动处理业务数据在不同表或数据库的分区及源表的 schema 变化。
  3. 新连接器:引入了新的 Apache Kafka 和 Apache Paimon 的 Pipeline Sink,增强了生态系统的扩展性,其中Kafka Sink 使得用户可以发送原始Debeizum/ Canal Json 格式的CDC数据到消息队列,Paimon Sink 则是让用户可以简单高效地完成MySQL实时入湖。
  4. 连接器改进:如 MySQL 增加了 tables.exclude 选项和 MysqlDebeziumTimeConverter,OceanBase 支持 DebeziumDeserializationSchema,Db2 迁移到统一增量快照框架等。

Flink CDC 3.1 核心特性解读

Transformation 支持

Flink CDC 3.1.0 引入了在 CDC pipeline 中进行数据变换(transformation)的功能。通过在 YAML pipeline 定义中加入 transform 部分,用户现在可以轻松地对来源的数据变更事件应用各种变换,包括投影、计算和添加常量列,从而提高数据集成管道的效率。新特性利用类似 SQL 的语法定义这些转换,确保用户可以快速适应并使用它。例如,只需编写如下 YAML 语句块:

transform:
  - source-table: db.tbl1
    projection: id, age, weight, height, weight / (height * height) as bmi
    filter: age > 18 AND name IS NOT NULL

即可对传递的数据流应用投影操作(仅保留原表中的部分列)、计算操作(根据原列数据计算新列并追加到数据记录中)和过滤操作(从结果中清除符合条件的数据行)。可以书写多条 Transform 规则,它们会同时生效。[1]

分库分表合并支持

Flink CDC 3.1.0 现在通过在 YAML pipeline 定义中配置 route,在分库分表场景下将多表合并为一个。由于业务数据量庞大,业务数据经常会被分别存放在多个表甚至数据库中。通过配置route,用户可以将多张源表映射至同一个目标表,在同步时,数据变更事件(DataChangeEvent)和 Schema 变更事件都将被合并到指定的目标表中。例如,只需编写如下 YAML 语句块:

route:
  - source-table: db.tbl\.*
    sink-table: db.unified
  - source-table: db.tbl_log\.*
    sink-table: db.log

即可将源库中所有匹配 tbl.*tbl_log.*正则表达式的分片表合并,并分别同步到下游的 db.unifieddb.log汇表中。(.用于分隔数据库名称和表名称,因此作为正则表达式关键字时需要使用 \进行转义。)可以书写多条 Route 规则,它们会同时生效。[2]

Flink CDC 3.1 新功能最佳实践

使用 Kafka Pipeline Sink 高效写入 Canal/Debezium 格式数据

Flink CDC 3.1.0 引入了新的 Kafka Pipeline Sink(基于 Kafka 3.2.3 版本)。现在,您可以编写如下所示的 YAML 语句块来定义一个从 MySQL 捕获变化数据并写入下游 Kafka Sink 的 Pipeline 作业[3]:

source:
   type: mysql
   # ...

sink:
  type: kafka
  properties.bootstrap.servers: PLAINTEXT://localhost:62510
  value.format: canal-json

该作业将来自 MySQL 上游的变化数据编码为 Canal JSON 格式,并写入到指定的 Kafka 服务器中;相比于 Flink SQL Changelog 格式,Flink CDC 不会将数据更新事件拆分为 BEFORE 和 AFTER 两条记录,能够更高效地处理分区表场景,并支持将事件序列化为 Debezium 和 Canal JSON 格式。

Flink 支持将上述格式解析为标准变更消息处理[4],因此您可以简单地使用以下 Flink SQL 将其摄入流式处理框架,整个过程无需额外部署 Canal 或 Debezium 集群,直接复用已有 Flink 集群即可:

CREATE TABLE topic_products (
  -- 上游的 Schema 结构
) WITH (
 'connector' = 'kafka',
  -- ...
 'properties.bootstrap.servers' = 'localhost:9092',
 'format' = 'canal-json'  -- 从 Kafka 摄取 Canal JSON 格式数据
)

完整的数据流示意图如下所示:

更高效地实时入湖 Paimon

Flink CDC 3.1.0 引入了新的 Apache Paimon Pipeline Sink(基于 Paimon 0.7.0 版本)。现在,您可以编写如下所示的 YAML 语句块来定义一个从 MySQL 捕获变化数据并写入下游 Paimon Sink 的 Pipeline 作业[5]:

source:
  type: mysql
  # ...

sink:
  type: paimon
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

可选择的下游元数据存储支持 FileSystem 和 Hive。在启用 Schema Evolution 选项时,Flink CDC 会同时捕获数据变更和表结构变更、在应用 Transform 和 Route 规则后将数据发送到下游,并将结构变更应用到 Catalog 中。完整的数据流示意图如下所示:

相比于使用 Flink SQL 和 Paimon CDC Action 的同步方案,Flink CDC Pipeline 作业支持将上游表结构变更动态应用至下游,且进一步支持了对上游表进行列投影和行过滤,提供细粒度的数据路由规则,追加计算列的逻辑配置更加简单。例如,以下 Paimon Action 变换语句[6]:

flink run paimon-action.jar
    --metadata_column "table_name"
    --computed_column "name=UPPER(name)"
    --computed_column "nameage=CONCAT(name, age)"

可以使用 Flink CDC YAML 等效地表述为:

projection: \*, __table_name__, UPPER(name) as newage, CONCAT(name, age) as nameage

其他改进

MySQL Pipeline 连接器

在此版本中,MySQL pipeline source 引入了一个新的选项 tables.exclude,用户可以更简单地使用正则表达式排除不必要的表。

MySQL Source 连接器

MySQL CDC source 同时新增了一个自定义转换器 MysqlDebeziumTimeConverter,用于将时间类型列转换为更易于读取和序列化的字符串。

OceanBase Source 连接器

OceanBase CDC source 现在支持指定通用的 DebeziumDeserializationSchema,以重用现有的 Debezium 反序列化器。

Db2 Source 连接器

Db2 CDC source 已经迁移至统一的增量快照框架。

SinkFunction 支持

尽管 SinkFunction 在 Flink 中已被标记为弃用,但考虑到一些 Flink connector 仍在使用该 API,我们也为 CDC pipeline sink 支持 SinkFunction API 以帮助扩展 Flink CDC 的生态系统。

CLI 支持从 savepoint 恢复

Flink CDC pipeline 提交 CLI 现在支持通过命令行参数 --from-savepoint 从特定的 savepoint 文件恢复 Flink 作业。

Flink CDC 3.1 版本兼容性

捐赠给 Apache 基金会使得 Flink CDC 项目更中立的同时也带来了短期的不便,因为 Apache 基金会对所属项目的包名、license 有着严格的要求,因此在 Flink CDC 3.1 版本之前版本存在不兼容的情况,我们在此详细说明。当然,后续的 3.2、3.3 版本会与 3.1 版本保持兼容。

Group ID 和 Package 路径变更

如果您正通过 Maven 或 Gradle 声明 Flink CDC 依赖,则需要在升级到 3.1 版本的同时将 Group ID 从 com.ververica.cdc改为 org.apache.flink.cdc,同时更改源代码中 import Package 路径。

用于 Flink SQL 作业的 Flink Source 连接器的重要更改

由于许可证与 Apache 2.0 License 不兼容,我们无法将以下连接器的 JDBC driver 包含在我们的二进制发布包中:

  • Db2
  • MySQL
  • Oracle
  • OceanBase

请手动将相应的 JDBC 驱动程序下载到 Flink 集群的 $FLINK_HOME/lib 目录中,或在使用 --jar 提交 YAML pipeline 时指定驱动程序的路径。如果您在使用 Flink SQL,请确保它们在 classpath 下。

作业 State 兼容性

由于以上不兼容的变更,使用 Flink CDC 3.1 以前版本保存的作业 State 无法在较新版本上恢复。因此,您需要在升级 Flink CDC 版本后进行一次无状态重启。

致谢

衷心感谢以下开发者为 Flink CDC 3.1.0 版本作出的贡献!

Check Null, FocusComputing, GOODBOY008, Hang Ruan, He Wang, Hongshun Wang, Jiabao Sun, Kunni, L, Laffery, Leonard Xu, Muhammet Orazov, Paul Lin, PengFei Li, Qingsheng Ren, Qishang Zhong, Shawn Huang, Thorne, TorinJie, Xianxun Ye, Xin Gong, Yaroslav Tkachenko, e-mhui, gongzhongqiang, joyCurry30, kunni, lzshlzsh, qwding, shikai93, sky, skylines, wenmo, wudi, xleoken, xuzifu666, yanghuaiGit, yuxiqian, 张田

参考链接:

[1] https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1/docs/core-concept/transform/

[2] https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1/docs/core-concept/route/

[3] https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1/docs/connectors/kafka/

[4] https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/connectors/table/formats/canal/

[5] https://nightlies.apache.org/flink/flink-cdc-docs-release-3.1/docs/connectors/paimon/

[6] https://paimon.apache.org/docs/master/flink/cdc-ingestion/mysql-cdc/


更多内容

img


活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算 Flink 版现开启活动:
新用户复制点击下方链接或者扫描二维码即可0元免费试用 Flink + Paimon
实时计算 Flink 版(3000CU*小时,3 个月内)
了解活动详情:https://free.aliyun.com/?pipCode=sc

0CA9E977-9C4C-4444-94B3-F01C0B8C891B.png

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
6天前
|
SQL 监控 关系型数据库
实时计算 Flink版产品使用问题之使用mysql cdc配置StartupOptions.initial()全量之后就不增量了,是什么原因
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
4天前
|
消息中间件 Oracle 关系型数据库
实时计算 Flink版操作报错合集之连接RabbitMQ时遇到Could not find any factory for identifier 'rabbitmq' that implements 'org.apache.flink.table.factories.DynamicTableFactory'错误,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
255 0
|
4天前
|
SQL 关系型数据库 MySQL
实时计算 Flink版操作报错合集之CDC任务在异常后整个record sent从0初始化开始,是什么导致的
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
362 0
|
4天前
|
关系型数据库 数据库 流计算
实时计算 Flink版操作报错合集之在使用Flink CDC TiDB Connector时,无法获取到事件,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
308 0
|
4天前
|
Kubernetes Oracle 关系型数据库
实时计算 Flink版操作报错合集之用dinky在k8s上提交作业,会报错:Caused by: org.apache.flink.table.api.ValidationException:,是什么原因
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
4天前
|
关系型数据库 MySQL 数据库
实时计算 Flink版操作报错合集之下载了mysql的cdc的demo,在本地调试时,报错:找不到这个包,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
4天前
|
消息中间件 Java Kafka
实时计算 Flink版操作报错合集之从hudi读数据,报错NoSuchMethodError:org.apache.hudi.format.cow.vector.reader.PaequetColumnarRowSplit.getRecord(),该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
4天前
|
Oracle 关系型数据库 Shell
实时计算 Flink版操作报错合集之遇到报错:Error: Could not find or load main class org.apache.flink.cdc.cli.CliFrontend,是什么导致的
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
4天前
|
消息中间件 Java 关系型数据库
实时计算 Flink版操作报错合集之从 PostgreSQL 读取数据并写入 Kafka 时,遇到 "initial slot snapshot too large" 的错误,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
551 0
|
4天前
|
存储 SQL 关系型数据库
实时计算 Flink版操作报错合集之按时间恢复时,报错:在尝试读取binlog时发现所需的binlog位置不再可用,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
487 0

热门文章

最新文章

相关产品

  • 实时计算 Flink版
  • 推荐镜像

    更多