开发者社区 > 大数据与机器学习 > 实时计算 Flink > 正文

Flink CDC中经过测试,好像没生效,修改其他列 CDC也能收到update流。什么原因?

"问题1:Flink CDC中经过测试,好像没生效,修改其他列 CDC也能收到update流。什么原因?
74bf75c0e52e5c9f63143a3622ab572c.png
'debezium.column.include.list' = 'wit_account_3.accountinfo[0-9]+.project_id,wit_account_3.accountinfo[0-9]+.account_id,wit_account_3.accountinfo[0-9]+.user_id,wit_account_3.accountinfo[0-9]+.use_code_password'
只加一个字段的话,其他字段值为null 对表的所有update操作都能收到数据流。
问题2:好像还不行,我用的tablesink,我看能不能 在sink中存 before状态数据,
然后after 数据来了做对比 感觉不太行, 用的是DynamicTableSink。
CREATE TABLE account_code_source
(
project_id bigint,
account_id bigint,
user_id bigint COMMENT '用户id',
use_code_password varchar(255),
write_time AS PROCTIME(),
PRIMARY KEY (project_id, account_id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'server-id' = '5702',
'hostname' = 'rRRRRRR,
'port' = '3306',
'username' = 'rRRRRRR',
'password' = 'rRRRRRR',
'database-name' = 'wit_account_3',
'table-name' = 'accountinfo[0-9]+',
'debezium.column.include.list' = 'wit_account_3.accountinfo[0-9]+.project_id,wit_account_3.accountinfo[0-9]+.account_id,wit_account_3.accountinfo[0-9]+.user_id,wit_account_3.accountinfo[0-9]+.use_code_password',
'scan.snapshot.fetch.size' = '2000'
);
SQL操作任务和数据,自定义了一个redis sink。用的是Java 自定义sink , invoke(RowData value, Context context)
有一个Context 对象 我以为可以拿到上下文数据。"

展开
收起
十一0204 2023-07-26 07:12:42 125 0
3 条回答
写回答
取消 提交回答
  • 如果在 Flink CDC 中进行测试后发现某个列的更新没有生效,但修改其他列时却能够收到 update 流,可能是以下原因之一:

    1. CDC 配置错误:请确保你正确配置了 Flink CDC 的连接器,包括源表和目标表的信息、数据类型映射、过滤条件等。检查配置是否准确且与实际需求相匹配。

    2. 源表结构变化:如果你修改了某个列的值,但没有触发 CDC 更新流,则可能是因为该列被排除在 CDC 监视的范围之外。检查 Flink CDC 的配置,确保修改的列已经正确地包含在 CDC 监视中。

    3. CDC Offset 管理问题:Flink CDC 使用 offset 来跟踪已处理的数据,并从上次停止的位置继续捕获数据。如果 offset 管理不正确,可能会导致无法捕获特定列的更新。检查 offset 管理机制是否正常工作,以确保每个字段的更新都可以被捕获。

    4. 版本或 bug 问题:Flink CDC 可能存在已知问题或 bug,导致特定列的更新无法被正确捕获。建议升级到最新版本,以获取更好的稳定性和功能。

    5. 其他问题:还有一些其他因素可能导致某个列的更新无法生效,如数据过滤条件、连接器配置错误、数据库权限问题等。请仔细检查相关设置并排除其他可能性。

    2023-07-31 23:35:24
    赞同 展开评论 打赏
  • 北京阿里云ACE会长

    如果在 Flink CDC 中修改了一列的 CDC 配置,但是其他列的 CDC 也能收到 update 流,可能是由于以下几个原因:
    Flink CDC 配置错误:请确认您的 Flink CDC 配置文件中,修改的列是否正确配置了 CDC 参数。如果配置错误,可能会导致其他列的 CDC 也能收到 update 流。
    源表结构变化:请确认您要同步的源表结构是否发生了变化。如果源表结构发生了变化,可能会导致 CDC 配置失效,从而出现其他列的 CDC 也能收到 update 流的情况。
    CDC 实现方式不一致:请确认您要同步的源表和目标表之间的 CDC 实现方式是否一致。如果源表和目标表之间的 CDC 实现方式不一致,可能会导致 CDC 配置失效,从而出现其他列的 CDC 也能收到 update 流的情况。
    数据库版本不兼容:请确认您使用的数据库版本是否与 Flink CDC 兼容。如果数据库版本不兼容,可能会导致 CDC 配置失效,从而出现其他列的 CDC 也能收到 update 流的情况。

    2023-07-29 17:09:36
    赞同 展开评论 打赏
  • 意中人就是我呀!

    回答1:你只能用datastream代码来比了。也可以和目标表来比,就写sql就行,目标表当作维表来处理。此回答整理至钉群“Flink CDC 社区”。

    2023-07-26 12:02:57
    赞同 展开评论 打赏

实时计算Flink版是阿里云提供的全托管Serverless Flink云服务,基于 Apache Flink 构建的企业级、高性能实时大数据处理系统。提供全托管版 Flink 集群和引擎,提高作业开发运维效率。

相关产品

  • 实时计算 Flink版
  • 相关电子书

    更多
    Flink CDC Meetup PPT - 龚中强 立即下载
    Flink CDC Meetup PPT - 王赫 立即下载
    Flink CDC Meetup PPT - 覃立辉 立即下载