flink cdc mysql

// Flink CDC 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.formats.json.JsonRowDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
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 {
        // 创建执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 MySQL CDC Source
        DataStream<String> mysqlSource = MySqlSource.<String>builder()
                .hostname("localhost")
                .port(3306)
                .databaseList("my_database") // 监听的数据库名称
                .tableList("my_database.my_table") // 监听的表名称
                .username("root")
                .password("password")
                .deserializer(new DebeziumDeserializationSchema<String>() {
                    private final ObjectMapper objectMapper = new ObjectMapper();

                    @Override
                    public void deserialize(JsonNode jsonNode, Collector<String> collector) throws Exception {
                        // 将 Debezium 的 JSON 格式转换为字符串
                        collector.collect(objectMapper.writeValueAsString(jsonNode));
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return TypeInformation.of(String.class);
                    }
                })
                .build();

        // 将数据写入到 Kafka 或其他目标系统
        SingleOutputStreamOperator<String> processedStream = mysqlSource
                .filter(record -> !record.isEmpty()) // 简单过滤掉空记录
                .map(record -> {
                    // 处理每条记录，例如解析 JSON 字符串
                    return record;
                });

        // 输出到控制台或 Sink 到其他系统
        processedStream.print();

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

解释说明：

1. 创建执行环境：使用 StreamExecutionEnvironment.getExecutionEnvironment() 创建一个 Flink 流处理环境。
2. 配置 MySQL CDC Source：通过 MySqlSource.builder() 构建 MySQL CDC 数据源，指定主机名、端口、数据库名、表名、用户名和密码。deserializer 用于将 Debezium 格式的 JSON 数据转换为字符串。
3. 处理数据流：对从 MySQL 捕获的数据进行简单过滤和处理，确保数据有效。
4. 输出数据：使用 print() 方法将处理后的数据输出到控制台，也可以将其 Sink 到其他系统（如 Kafka）。
5. 执行 Flink 作业：调用 env.execute() 启动 Flink 作业。

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