Write the changed data in MySQL to Kafka through flinkcdc (datastream mode)

---

Preface

CDC yes Change Data Capture（ Change data capture ） Abbreviation

FlinkCDC The core idea of is to monitor and capture changes in the database （ Includes the insertion of data or data tables 、 Update and delete, etc ）, Keep a complete record of these changes in the order they occur , Write to message middleware for other services to subscribe and consume .

---

One 、CDC The type of ？

CDC It is mainly divided into query based and Binlog Two ways , Briefly describe the difference between the two ：

	Query based CDC	be based on binlog Of CDC
Common components	Sqoop	Maxwell、Canal、Debezium
thought	Batch	Streaming
Delay	high	low
Whether all data changes can be captured	no	yes

Flink The community has developed flink-cdc-connectors Components , This is one that can be directly from MySQL、PostgreSQL The database directly reads the total data and incremental change data source Components .

Two 、 adopt FlinkCDC Take data from MySQL Import to Kafka

1. Core code

：

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.app.function.CustomerDeserialization;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


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

        //1. Get execution environment 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.flinkcdc structure SourceFunction
        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop101")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("gmall-flink")
                .tableList("gmall-flink.base_trademark")
                .deserializer(new CustomerDeserialization())
                .startupOptions(StartupOptions.latest())
                .build();
        DataStreamSource<String> streamSource = env.addSource(sourceFunction);

        //3. Print data and write data kafka
        streamSource.print();
        String sinkTopic = "ods_base_db";
        streamSource.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

        //4. Start the task 
        env.execute("FlinkCDC");

    }
}

2. Tool class

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

public class MyKafkaUtil {
    

    public static FlinkKafkaProducer<String> getKafkaProducer(String topic){
    

        return new FlinkKafkaProducer<String>("192.168.2.101:9092", topic, new SimpleStringSchema());
    }
}

---

3、 ... and 、 Result display

I am here MySQL Every piece of data changed in the database , stay IDEA and kafka It can be detected here ：
IDEA:

Kafka：

3. Possible errors and solutions

org.apache.kafka.common.errors.TimeoutException: Topic ods_base_database not present in metadata after 60000 ms

solve ：

1.vi kafka/config.server.properties

Modify these three places , Remember to use IP Address , Previously used hadoop101 Keep making mistakes

2. restart kafka, restart zookeeper, Problem solving ！