目录

1.写在前面

2.Maven依赖

3.代码实现-普通实现

4.集群测试

4.1 环境准备

4.2 查看任务结果

1.写在前面

        Flink CDC有两种实现方式，一种是DataStream，另一种是FlinkSQL方式。

DataStream方式：优点是可以应用于多库多表，缺点是需要自定义反序列化器（灵活）FlinkSQL方式：优点是不需要自定义反序列化器，缺点是只能应用于单表查询

2.Maven依赖

        
            org.apache.flink
            flink-java
            1.12.7
        

        
            org.apache.flink
            flink-streaming-java_2.12
            1.12.7
        

        
            org.apache.flink
            flink-clients_2.12
            1.12.7
        

        
            org.apache.hadoop
            hadoop-client
            2.7.7
        

        
            mysql
            mysql-connector-java
            5.1.49
        

        
            com.alibaba.ververica
            flink-connector-mysql-cdc
            1.2.0
        

        
            com.alibaba
            fastjson
            1.2.75
        

        
            org.apache.flink
            flink-table-planner-blink_2.12
            1.12.7
        
    

    
        
            
                org.apache.maven.plugins
                maven-assembly-plugin
                3.0.0
                
                    
                        jar-with-dependencies
                    
                
                
                    
                        make-assembly
                        package
                        
                            single
                        
                    
                
            
            
                org.apache.maven.plugins
                maven-compiler-plugin
                
                    8
                    8
                
            
        
    

3.代码实现

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class FlinkCDCWithSQL {

    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2.DDL方式建表,flink_sql的方式只能监控一张表
        tableEnv.executeSql("CREATE TABLE mysql_binlog ( " +
                " id STRING NOT NULL, " +
                " tm_name STRING, " +
                " logo_url STRING " +
                ") WITH ( " +
                " 'connector' = 'mysql-cdc', " +
                " 'hostname' = '192.168.0.111', " +
                " 'port' = '3306', " +
                " 'username' = 'root', " +
                " 'password' = '123456', " +
                " 'database-name' = 'gmall2021', " +
                " 'table-name' = 'base_trademark' " +
                ")");

        //3.查询数据
        Table table = tableEnv.sqlQuery("select * from mysql_binlog");

        //4.将动态表转换为流
        DataStream> retractStream = tableEnv.toRetractStream(table, Row.class);
        retractStream.print();

        //5.启动任务
        env.execute("FlinkCDCWithSQL");

    }

}