flink cdc踩坑日记

最近几天想玩一下flink cdc，然而出现了神奇的一幕：有主键的表可以正常连接正常捕捉变化数据，但是没有主键的表却迟迟没有输出

1. flink cdc代码实例

maven依赖

<properties>
    <scala.version>2.11</scala.version>
    <flink.version>1.13.6</flink.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-core</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <dependency>
        <groupId>com.ververica</groupId>
        <artifactId>flink-connector-mysql-cdc</artifactId>
        <version>2.2.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-api-scala-bridge_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

核心代码：

public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("ubuntu")
            .port(3306)
            .databaseList("testdb") // 设置数据库，可以填写多个
            .tableList("testdb.products") // 设置要cdc的表，可以设置多个
            .username("root")
            .password("123456")
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .build();

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // enable checkpoint
    env.enableCheckpointing(3000);

    env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
            // set 4 parallel source tasks
            .setParallelism(4)
            .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

    env.execute("Print MySQL Snapshot + Binlog");
}

product表如下：

连接之后可以正常输出，美汁汁：

但是当我要cdc另一个没有主键的表时，迟迟捕捉不到！妈蛋我分析了半天原因，最后定位到主键这个问题。

---

2. 聊聊遇到的问题

为啥没有主键就没办法cdc到了呢？
我说说自己的理解：
我这里开启的是mysql的row模式，众所周知，mysql的binlog有三种模式：statement、row和mixed：

• statement模式下binlog只顺序记录数据库中执行的语句，从库在使用binlog进行备份时，读取这些语句并且顺序执行，以获得与主库同样的数据。但是这样做有个问题，例如我们在sql语句中使用了rand函数，那么在主库和从库中产生的结果可能是不同的，因此使用statement模式可能会导致主从的不一致。
• 为了解决这一问题，row模式在binlog中记录主库在执行一条语句前后的数据。使用这种方式，从库可以获取与主库完全一致的状态，但是这样做的弊端也是很明显的，那就是数据文件太大，我们可能只是执行了一条sql语句，就可能导致要记录非常多的数据变化。
• mixed：这种模式如名字所示，结合了statement模式和row模式，在执行确定的语句时在binlog中使用statement模式记录sql语句，在执行不确定语句时使用row模式记录状态变化；虽然想法非常棒，但是还是有可能有意外情况法会导致主从不一致。

我开启的是row模式，这直接导致了本文出现的问题。
我们知道binlog设计出来是为了进行mysql的主从备份的，假设说我修改了一行数据，为了将这个变化记录到binlog中，我们最直接的想法是啥呢？

如果是我的话，我设计的数据结构一定长这个样：

----------------------------------------------------
 库  |  表  |  主键   |   before state | after state|
----------------------------------------------------

因为mysql是使用主键建立索引的，当从库在同步时，直接读取发生变化的主键，快速找到那条数据，然后讲话数据的状态从before state更改为after state。

那没有主键咋整呢？
这个我没有精确地查证mysql是怎么做的，我认为没有主键是无法进行状态同步的，因为如果没有主键，主库中可能具有完全相同的很多行，在我们修改了其中一行的时候，无法将“我修改了哪一行”这条信息准确地传达给从库，因此从库无法进行同步。