FlinkSql cdc mysql to mysql

pom文件

 
        1.13.6
        2.12
    

    
        
            org.apache.flink
            flink-connector-jdbc_${scala.binary.version}
            ${flink.version}
        
        
            org.apache.flink
            flink-table-planner-blink_${scala.binary.version}
            ${flink.version}
        
        
            org.apache.flink
            flink-java
            ${flink.version}
        
        
            org.apache.flink
            flink-streaming-java_${scala.binary.version}
            ${flink.version}
        
        
            org.apache.flink
            flink-clients_${scala.binary.version}
            ${flink.version}
        
        
            org.apache.flink
            flink-runtime-web_${scala.binary.version}
            ${flink.version}
        
        
            com.alibaba.ververica
            flink-connector-mysql-cdc
            1.4.0
        

        
            org.projectlombok
            lombok
            1.18.20
        

        
            com.alibaba
            fastjson
            1.2.75
        
    

    
        
            
                org.apache.maven.plugins
                maven-shade-plugin
                3.1.0
                
                    false
                
                
                    
                        package
                        
                            shade
                        

                        
                            
                                
                                    
                                    demo.flinkcdc_binlog.MysqlBinlogCdcToMysql
                                
                                
                                    reference.conf
                                
                            
                            
                                
                                    *:*:*:*
                                    
                                        META-INF/*.SF
                                        META-INF/*.DSA
                                        META-INF/*.RSA
                                    
                                
                            
                        
                    
                
            
            
                org.apache.maven.plugins
                maven-compiler-plugin
                
                    6
                    6
                
            
        
    

代码

package flinksql;

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

public class flinksql_cdc {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql("create table demo1(" +
                " id INT," +
                " username STRING," +
                " password STRING," +
                " primary key (id) not enforced" +
                ")WITH(" +
                " 'connector' = 'mysql-cdc'," +
                " 'hostname' = 'localhost'," +
                " 'port' = '3306'," +
                " 'username' = 'root'," +
                " 'password' = '123456'," +
                " 'database-name' = 'old_data'," +
                " 'table-name' = 'demo'" +
                ")");

        tableEnv.executeSql("create table demo2(" +
                " id INT," +
                " username STRING," +
                " password STRING," +
                " primary key (id) not enforced" +
                ")WITH(" +
                " 'connector' = 'jdbc'," +
                " 'url' = 'jdbc:mysql://localhost:3306/new_data?serverTimezone=UTC'," +
                " 'username' = 'root'," +
                " 'password' = '123456'," +
                " 'table-name' = 'demo_info'," +
                " 'driver' = 'com.mysql.cj.jdbc.Driver'," +
                " 'scan.fetch-size' = '200'" +
                ")");
        tableEnv.executeSql("insert into demo2 select id,username,password from demo1");
        tableEnv.executeSql("select * from demo1").print();
        env.execute("FLinkSql Cdc");

    }
}

上面是增删改的效果图，可以根据 “op” 查看具体操作
I是插入，U是更新，D是删除
-U是撤回前的数据，+U是更新后的数据
可以结合数据库中的数据判断是否同步成功

注意：flinksql和mysql的表都需要设置主键