Flink CDC 实战：监听数据库表变化并打印到控制台

使用 Flink CDC 来监听表的增删改数据并打印到控制台，需要遵循以下步骤：

首先，您需要在 Flink 作业中添加 Flink CDC 依赖项。可以通过在 pom.xml 文件中添加以下依赖项来完成此操作：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka_2.12</artifactId>
  <version>${flink.version}</version>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-sql-connector-kafka</artifactId>
  <version>${flink.version}</version>
</dependency>

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-jdbc_2.12</artifactId>
  <version>${flink.version}</version>
</dependency>

然后，您需要创建一个 Flink CDC 源。可以使用以下代码创建它：

public class FlinkCDCSource {
    private static final String driverName = "com.mysql.jdbc.Driver";
    private static final String url = "jdbc:mysql://localhost:3306/test";
    private static final String username = "root";
    private static final String password = "password";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        JdbcConnectionOptions jdbcConnectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(url)
                .withDriverName(driverName)
                .withUsername(username)
                .withPassword(password)
                .build();

        DebeziumDeserializationSchema<RowData> deserializer = DebeziumDeserializationSchema.forRowData();
        FlinkDebeziumConsumer<RowData> consumer = new FlinkDebeziumConsumer<>(jdbcConnectionOptions, deserializer);

        DataStreamSource<RowData> source = env.addSource(consumer);
        source.print();

        env.execute("FlinkCDCSource");
    }
}

在此示例中，我们使用 JDBC 连接选项来连接到 MySQL 数据库，并使用 Debezium 反序列化模式为 RowData 创建一个 DebeziumDeserializationSchema。然后，我们将 FlinkDebeziumConsumer 添加到数据源中，并打印每个事件。

最后，您需要使用以下命令运行作业：

./bin/flink run --classpath file:///path/to/flink-cdc.jar /path/to/flink-cdc.jar

在此命令中，您需要将 flink-cdc.jar 替换为您的 Flink CDC 源代码的 JAR 文件的路径。然后，Flink 将启动作业，并开始监听表的增删改数据。每当发生更改时，它将打印事件到控制台。