Flink CDC 连接器使用教程

项目目录结构及介绍

Flink CDC 连接器项目的目录结构如下：

flink-cdc-connectors/
├── flink-cdc-base
│   ├── src
│   │   ├── main
│   │   │   ├── java
│   │   │   │   ├── com
│   │   │   │   │   ├── ververica
│   │   │   │   │   │   ├── flinkcdc
│   │   │   │   │   │   │   ├── base
│   │   │   │   │   │   │   ├── debezium
│   │   │   │   │   │   │   ├── mysql
│   │   │   │   │   │   │   ├── oracle
│   │   │   │   │   │   │   ├── postgres
│   │   │   │   │   │   │   ├── sqlserver
│   │   │   │   │   │   │   ├── util
│   │   │   │   │   │   │   └── ...
│   │   │   ├── resources
│   │   │   └── ...
│   │   └── test
│   │       ├── java
│   │       │   └── ...
│   │       └── resources
│   │           └── ...
│   └── pom.xml
├── flink-cdc-examples
│   ├── src
│   │   ├── main
│   │   │   ├── java
│   │   │   │   └── ...
│   │   │   └── resources
│   │   │       └── ...
│   │   └── test
│   │       ├── java
│   │       │   └── ...
│   │       └── resources
│   │           └── ...
│   └── pom.xml
├── flink-cdc-parent
│   └── pom.xml
├── .gitignore
├── LICENSE
├── NOTICE
├── README.md
└── pom.xml

主要目录介绍

• flink-cdc-base: 包含 Flink CDC 连接器的基础实现和各个数据库的连接器实现。
• flink-cdc-examples: 包含使用 Flink CDC 连接器的示例代码。
• flink-cdc-parent: 项目的父模块，管理依赖和插件。

项目的启动文件介绍

Flink CDC 连接器的启动文件通常位于 flink-cdc-examples 目录下。以下是一个典型的启动文件示例：

package com.ververica.flinkcdc.examples;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.flinkcdc.debezium.DebeziumSourceFunction;
import com.ververica.flinkcdc.debezium.DebeziumDeserializationSchema;

public class MySqlCDCExample {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DebeziumSourceFunction<String> sourceFunction = DebeziumSourceFunction.builder()
            .hostname("localhost")
            .port(3306)
            .database("your_database")
            .table("your_table")
            .username("your_username")
            .password("your_password")
            .deserializer(new DebeziumDeserializationSchema<String>() {
                @Override
                public void deserialize(byte[] message, Collector<String> out) throws Exception {
                    out.collect(new String(message));
                }
            })
            .build();

        env.addSource(sourceFunction).print();

        env.execute("Flink MySQL CDC Example");
    }
}

启动文件关键点

• StreamExecutionEnvironment: 创建 Flink 流执行环境。
• DebeziumSourceFunction: 配置 Debezium 源函数，包括数据库连接信息和反序列化器。
• env.execute: 启动 Flink 作业。

项目的配置文件介绍

Flink CDC 连接器的配置文件通常位于 flink-cdc-base/src/main/resources 目录下。以下是一个典型的配置文件示例：

# Debezium Connector Configuration
connector.class=io.debezium.connector.mysql.MySqlConnector
database.hostname=localhost
database.port=330