flink cdc入门案例

### Flink CDC入门示例教程

#### 准备工作
为了能够顺利运行Flink CDC项目，需先准备好环境配置。这包括安装并启动Apache Flink集群以及确保MySQL数据库处于可用状态[^1]。

#### 创建Maven工程
创建一个新的Maven工程来构建Flink应用，在`pom.xml`文件中加入必要的依赖项用于支持Flink流处理和连接器功能：

<dependencies>
    <!-- Apache Flink dependencies -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- MySQL CDC connector dependency -->
    <dependency>
        <groupId>com.ververica</groupId>
        <artifactId>flink-connector-mysql-cdc</artifactId>
        <version>${cdc.connector.version}</version>
    </dependency>
</dependencies>

#### 编写Flink CDC程序
下面是一个简单的Java代码片段展示如何设置从MySQL读取变化的数据源，并打印这些记录到控制台:

import com.alibaba.ververica.cdc.connectors.mysql.MySqlSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("localhost") // 数据库主机地址
                .port(3306)            // 数据库端口
                .databaseList("testdb")// 要监控的数据库名列表
                .tableList("testdb.tab1") // 表格名称, 支持正则表达式匹配多个表格
                .username("root")      // 用户名
                .password("password")  // 密码
                .deserializer(new SimpleStringDeserializer()) // 反序列化函数
                .build();

        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
           .print(); 

        env.execute("Print MySQL binlog change");
    }
}

这段代码定义了一个名为 `mySqlSource` 的数据源对象，它会监听指定的MySQL实例上的特定表的变化事件并将它们作为字符串输出至标准输出设备[^2]。