环境说明：

flink 1.15.2

mysql 版本5.7    注意：需要开启binlog，因为增量同步是基于binlog捕获数据

windows11 IDEA 本地运行

具体前提设置，请看这篇，包含 binlog 设置、Maven......

Flink CDC 基于mysql binlog 实时同步mysql表_彩虹豆的博客-CSDN博客

经过不懈努力，终于从阿里help页面找到了支持无主键同步的参数：

MySQL_实时计算 Flink版-阿里云帮助中心

 然后就开始一顿模式，各种参数调试，终于达到了目的，无主键表实时同步，只不过在sink表关联目标表时，要指定几个字段为主键，这样就不会有重复的覆盖情况了，多给几个字段作为主键，不就避免重复冲突了嘛。比如id+date+local等，具体看表字段。

demo如下：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class MysqlToMysqlNonePrimaryKey {

    public static void main(String[] args) {
        //1.获取stream的执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);
        EnvironmentSettings settings = EnvironmentSettings.newInstance().build();
        //2.创建表执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv, settings);

        String sourceTable = "CREATE TABLE mysql_cdc_source (" +
                "  id INT,
" +
                "  username STRING,
" +
                "  password STRING
" +
                ") WITH (
" +
                "'connector' = 'mysql-cdc',
" +
                "'hostname' = 'localhost',
" +
                "'port' = '3306',
" +
                "'username' = 'root',
" +
                "'password' = 'root',
" +
                "'database-name' = 'test_cdc',
" +
                "'debezium.snapshot.mode' = 'initial',
" +
                "'scan.incremental.snapshot.enabled' = 'false',
" +
                //如果开启增量快照，必须设置主键。
                //默认开启增量快照。增量快照是一种读取全量数据快照的新机制。与旧的快照读取相比，增量快照有很多优点，包括：
                //读取全量数据时，Source可以是并行读取。
                //读取全量数据时，Source支持chunk粒度的检查点。
                //读取全量数据时，Source不需要获取全局读锁（FLUSH TABLES WITH read lock）。
                //如果您希望Source支持并发读取，每个并发的Reader需要有一个唯一的服务器ID，因此server-id必须是5400-6400这样的范围，并且范围必须大于等于并发数。
                "'scan.incremental.snapshot.chunk.key-column' = 'id' ,
" +
                //可以指定某一列作为快照阶段切分分片的切分列。无主键表必填，选择的列必须是非空类型（NOT NULL）。
                //有主键的表为选填，仅支持从主键中选择一列。
                "  'table-name' = 'user'
" +
                ")";
        tEnv.executeSql(sourceTable);
//        tEnv.executeSql("select * from mysql_cdc_source").print();
        String sinkTable = "CREATE TABLE mysql_cdc_sink (" +
                "  id INT,
" +
                "  username STRING,
" +
                "  password STRING
" +
                "  ,PRIMARY KEY (id,username,password) NOT ENFORCED
" +
                ") WITH (
" +
                "'connector' = 'jdbc',
" +
                "'driver' = 'com.mysql.cj.jdbc.Driver',
" +
                "'url' = 'jdbc:mysql://localhost:3306/" + "test_cdc" + "?rewriteBatchedStatements=true',
" +
                "'username' = 'root',
" +
                "'password' = 'root',
" +
                "'table-name' = 'user_new'
" +
                ")";
        tEnv.executeSql(sinkTable);
        tEnv.executeSql("insert into mysql_cdc_sink select id,username,password from mysql_cdc_source");
    }
}

由于无主键， debezium.snapshot.mode' = 'initial',这个参数会导致，程序运行几次，源表数据就会同步几次到目标表，并不会去重，如果想一直这个参数运行，需要在插入前先清空表，但是如果是数据量大的，推荐还是先用这个参数同步历史数据，完成后，再改为 schema_only，启动程序，然后把上面一个程序干掉。