使用flink1.13.5 mysql-cdc 向mysql增量同步数据的时候，为什么刚开始数据是准确稳定的，跑了一天就不稳定了，开始丢数据，丢字段信息

[18535916045]

val config: CheckpointConfig = env.getCheckpointConfig
env.enableCheckpointing(1000)
env.getCheckpointConfig.setCheckpointTimeout(60*1000)
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
env.getCheckpointConfig.setCheckpointStorage("hdfs://192.168.10.20:8020/flink/cms_product")
env.getCheckpointConfig.setTolerableCheckpointFailureNumber(3)

//读取product.product
tableEnv.executeSql(
  """
    |CREATE TABLE product_product (
    | id bigint,
    | merchant_product_id bigint,
    | code string,
    | channel_code string,
    | store_id bigint,
    | merchant_id bigint,
    | first_shelf_time timestamp,
    | type_of_product int,
    | can_sale int,
    | can_sale_type int,
    | ref_id bigint,
    | purchase_warehouse  int,
    | data_type int,
    | company_id bigint,
    | is_deleted bigint,
    | is_invalid int,
    | can_sale  int,
    |  price_level int,
    | parent_id bigint,
    | create_time timestamp,
    | update_time timestamp,
    | proctime as PROCTIME(),
    | PRIMARY KEY (id) NOT ENFORCED
    |) WITH (
    |    'connector' = 'mysql-cdc',
    |    'hostname' = '*',
    |    'port' = '3306',
    |    'username' = '*',
    |    'password' = '*',
    |    'database-name' = 'product',
    |    'table-name' = 'product',
    |    'server-time-zone'='Asia/Shanghai',
    |    'scan.startup.mode'='latest-offset'
    |)
    |""".stripMargin)



//读取product.product_info
tableEnv.executeSql(
  """
    |CREATE TABLE product_product_info (
    | id bigint,
    | use_type int,
    | brand_id bigint,
    | chinese_name string,
    | category_id bigint,
    | category_full_id_path string,
    | ext_field1 string,
    | product_channel string,
    | product_source string,
    | is_deleted int,
    | is_available int,
    | proctime as PROCTIME(),
    | PRIMARY KEY (id) NOT ENFORCED
    |) WITH (
    |    'connector' = 'mysql-cdc',
    |    'hostname' = '*',
    |    'port' = '3306',
    |    'username' = '*',
    |    'password' = '*',
    |    'database-name' = 'product',
    |    'table-name' = 'product_info',
    |    'server-time-zone'='Asia/Shanghai',
    |    'scan.startup.mode'='latest-offset'
    |)
    |""".stripMargin)

[xiaokai3006]

我也是同样的问题，我是很多表做左外连接打宽。刚开始的时候稳定，过几个小时就开始丢数据。flink 1.13.5，cdc 2.1.1

[18535916045]

你的checkpoint有没有报错失败，或者你这个问题有没有解决方案

…

------------------ 原始邮件 ------------------ 发件人: "ververica/flink-cdc-connectors" ***@***.***>; 发送时间: 2022年3月4日(星期五) 上午10:12 ***@***.***>; ***@***.******@***.***>; 主题: Re: [ververica/flink-cdc-connectors] 使用flink1.13.5 mysql-cdc 向mysql增量同步数据的时候，为什么刚开始数据是准确稳定的，跑了一天就不稳定了，开始丢数据，丢字段信息 (Discussion #896) 我也是同样的问题，我是很多表做左外连接打宽。刚开始的时候稳定，过几个小时就开始丢数据。flink 1.13.5，cdc 2.1.1 — Reply to this email directly, view it on GitHub, or unsubscribe. Triage notifications on the go with GitHub Mobile for iOS or Android. You are receiving this because you authored the thread.Message ID: ***@***.***>

[zhujian86]

你这么跑，内存占用怎么样？
看着用的是默认的内存状态后端，跑的时间长，把内存吃光了

[ctysf]

我也遇到同样的问题，数据开始跑是对的，跑着跑着，数据就丢了一部分，然后重启就好了，用的是mysql cdc

[yeezychao]

我也遇到了同样的问题在cdc 2.x版本，长时间运行后新增的binlog就不消费了，currentFetchEventTimeLag出现延迟，但source没有流量