flinkcdc mysql

// FlinkCDC MySQL 示例代码

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.connector.jdbc.catalog.JdbcCatalog;
import org.apache.flink.connector.jdbc.xa.XaDataSourceFactory;
import org.apache.flink.formats.json.JsonRowDeserializationSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.types.Row;
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;

import java.sql.PreparedStatement;
import java.util.Properties;

public class FlinkCDCMySQLExample {

    public static void main(String[] args) throws Exception {
        // 创建Flink流执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置MySQL CDC Source
        MySqlSource<JsonNode> mySqlSource = MySqlSource.<JsonNode>builder()
                .hostname("localhost")
                .port(3306)
                .databaseList("mydb") // 监听的数据库
                .tableList("mydb.mytable") // 监听的表
                .username("root")
                .password("password")
                .deserializer(new JsonDebeziumDeserializationSchema()) // 自定义反序列化器
                .build();

        // 从MySQL中读取数据
        DataStream<JsonNode> stream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source");

        // 处理数据（这里简单地将数据打印到控制台）
        stream.print();

        // 执行Flink作业
        env.execute("Flink CDC MySQL Example");
    }

    // 自定义Debezium反序列化器
    public static class JsonDebeziumDeserializationSchema implements DebeziumDeserializationSchema<JsonNode> {
        @Override
        public void deserialize(Envelope envelope, Collector<JsonNode> out) throws Exception {
            // 将Debezium的记录转换为JsonNode对象
            JsonNode jsonNode = objectMapper.convertValue(envelope, JsonNode.class);
            out.collect(jsonNode);
        }
    }
}

解释说明：

1. 创建Flink流执行环境：通过StreamExecutionEnvironment.getExecutionEnvironment()创建一个Flink的流处理环境。

2. 配置MySQL CDC Source：使用MySqlSource.builder()来构建一个MySQL CDC Source，指定监听的数据库和表，并设置用户名、密码等连接信息。这里使用了JsonDebeziumDeserializationSchema来自定义反序列化逻辑。

3. 从MySQL中读取数据：通过env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")将MySQL中的变更数据读取到Flink的数据流中。

4. 处理数据：这里简单地将数据打印到控制台，实际应用中可以根据需求进行更复杂的处理。

5. 执行Flink作业：通过env.execute("Flink CDC MySQL Example")启动Flink作业。

6. 自定义Debezium反序列化器：实现了DebeziumDeserializationSchema<JsonNode>接口，用于将Debezium的记录转换为JsonNode对象，便于后续处理。

这段代码展示了如何使用Flink CDC从MySQL中捕获变更数据并进行处理。