通过FlinkCDC将MySQL中变更的数据写入到kafka（DataStream方式）_flink cdc 写入kafka

---

前言

CDC是Change Data Capture（变更数据捕获）的缩写

FlinkCDC的核心思想是监测并捕获数据库的变动（包括数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。

---

一、CDC的种类？

CDC 主要分为基于查询和基于 Binlog 两种方式，简述两者的区别：

Flink 社区开发了 flink-cdc-connectors 组件，这是一个可以直接从 MySQL、PostgreSQL 等数据库直接读取全量数据和增量变更数据的 source 组件。

二、通过FlinkCDC将数据从MySQL导入到Kafka

1.核心代码

：

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.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.flinkcdc构建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.打印数据并将数据写入kafka
        streamSource.print();
        String sinkTopic = "ods_base_db";
        streamSource.addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

        //4.启动任务
        env.execute("FlinkCDC");

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

2.工具类

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());
    }
}
1
2
3
4
5
6
7
8
9
10

---

三、结果展示

我在MySQL数据库中每变更一条数据，在IDEA和kafka这边都可以检测到：
IDEA:

Kafka：

3.可能会出现的错误及解决

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

解决：

1.vi kafka/config.server.properties

修改这三个地方，切记用IP地址，之前用的hadoop101一直出错

2.重启kafka，重启zookeeper，问题解决！