flink通过cdc连接到tidb的代码

以下是连接 Flink 到 TiDB 使用 CDC 的代码示例：

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
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.cdc.JdbcCDCOptions;
import org.apache.flink.connector.jdbc.cdc.JdbcSource;
import org.apache.flink.connector.jdbc.cdc.JdbcSourceBuilder;
import org.apache.flink.connector.jdbc.cdc.JdbcSplit;
import org.apache.flink.connector.jdbc.cdc.TableId;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.types.Row;

import java.sql.Types;
import java.util.Properties;

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

        Properties properties = new Properties();
        properties.setProperty(JdbcConnectionOptions.URL.key(), "jdbc:mysql://tidb_ip:tidb_port/test");
        properties.setProperty(JdbcConnectionOptions.USERNAME.key(), "tidb_username");
        properties.setProperty(JdbcConnectionOptions.PASSWORD.key(), "tidb_password");

        JdbcCDCOptions cdcOptions = JdbcCDCOptions.builder()
                .setDatabaseList("test")
                .setTableList("test.user")
                .setServerTimeZone("Asia/Shanghai")
                .setParallelism(1)
                .build();

        TableId tableId = TableId.forDatabaseTable("test", "user");

        JdbcSourceBuilder sourceBuilder = JdbcSource.builder()
                .setOptions(properties)
                .setTableId(tableId)
                .setCDCOptions(cdcOptions);

        DataStream<Row> sourceStream = env.addSource(sourceBuilder.build());

        TypeInformation<Row> rowTypeInformation = new RowTypeInfo(Types.INTEGER, Types.VARCHAR, Types.INTEGER);

        SinkFunction<Row> sinkFunction = JdbcSink.sink(
                "INSERT INTO test.user_copy (id, name, age) VALUES (?, ?, ?)",
                new JdbcStatementBuilder<Row>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, Row row) throws SQLException {
                        preparedStatement.setInt(1, (Integer) row.getField(0));
                        preparedStatement.setString(2, (String) row.getField(1));
                        preparedStatement.setInt(3, (Integer) row.getField(2));
                    }
                },
                JdbcExecutionOptions.builder()
                    .withBatchSize(1000)
                    .withMaxRetries(5)
                    .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                    .withUrl("jdbc:mysql://tidb_ip:tidb_port/test")
                    .withDriverName("com.mysql.jdbc.Driver")
                    .withUsername("tidb_username")
                    .withPassword("tidb_password")
                    .build()
        );

        sourceStream.addSink(sinkFunction);

        env.execute("TiDB CDC Example");
    }
}

上面的代码示例中，我们首先定义了 TiDB 的连接信息和 CDC 配置，然后使用 `JdbcSourceBuilder` 构建了一个 CDC 的数据源。接着，我们定义了目标表的结构和插入语句，并使用 `JdbcSink` 将数据流写入到 TiDB 中。最后，将数据源和目标数据表连接起来，执行 Flink 作业即可。