代码

创建 java 类，如下：

package std;

import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class S1 {

    public static void main(String[] args) throws Exception {
        DebeziumSourceFunction<String> build = MySqlSource.<String>builder()
                .hostname("hadoop1")
                .port(3306)
                .databaseList("scott") // 设置要捕获数据的数据库，可传入多个数据库
                .tableList("scott.dept") // 设置要捕获数据的表，必须是 【库名.表名】的形式。如果不传，表示监控上面数据库中所有表的数据
                .username("root")
                .password("root")
                .deserializer(new StringDebeziumDeserializationSchema()) // 反序列化器
                .startupOptions(StartupOptions.initial())
                .build();

        // 获取flink环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);

        //env.enableCheckpointing(3000);

        // 添加数据源
        DataStreamSource<String> source = env.addSource(build);
        // 打印
        source.print();

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

解释：

• StartupOptions.initial())：将表中原有的数据打印到控制台，见 Flink CDC2.0 StartupOptions作用

执行结果

控制台打印一条记录很长，是 SourceRecord、ConnectRecord 对象的 toString() 结果，如下：

SourceRecord{
sourcePartition={server=mysql_binlog_source}, # mysql binlog
sourceOffset={
    ts_sec=1655552829, # binlog 时间戳
    file=mysql-bin.000004, # binlog 文件名
    pos=154,
    snapshot=true
    }
}

ConnectRecord{
  topic='mysql_binlog_source.scott.dept', # 表名
  kafkaPartition=null, 
  key=Struct{deptno=7369}, # 主键
  keySchema=Schema{mysql_binlog_source.scott.dept.Key:STRUCT},
  value=Struct{
  after=Struct{deptno=74,dname=财务部,loc=杭州2}, # 数据
  source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1655552829388,snapshot=true,db=scott,table=dept,server_id=0,file=mysql-bin.000004,pos=154,row=0},
  op=r, # crud中的 Retrieve，表示查询出的数据
  ts_ms=1655552829392
  }, valueSchema=Schema{mysql_binlog_source.scott.dept.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}

原文出处：/show_1IX3WLYT18Ze.html