Flink CDC我一开始就用代码，配置了一堆，没法使用代码玩了吗？

在Flink CDC中，如果你使用代码配置了Flink的表和数据流，那么你可以使用Flink的Table API或SQL API来创建和处理表和数据流。这些API提供了丰富的操作，可以帮助你创建、查询和处理表和数据流。
如果你想要将数据流写入到ClickHouse中，你可以使用Flink的Table API或SQL API来创建一个SinkFunction，并将这个SinkFunction应用到你的数据流中。这样，你就可以将数据流写入到ClickHouse中了。
如果你使用代码配置了Flink的表和数据流，那么你可以使用Flink的Table API或SQL API来创建和处理表和数据流。你可以使用TableEnvironment来创建和处理表和数据流，也可以使用DataStream和DataSet来处理数据流。这些API提供了丰富的操作，可以帮助你创建、查询和处理表和数据流。
如果你想要将数据流写入到ClickHouse中，你可以使用Flink的Table API或SQL API来创建一个SinkFunction，并将这个SinkFunction应用到你的数据流中。这样，你就可以将数据流写入到ClickHouse中了。

Flink CDC支持多种数据源和数据接收器，包括ClickHouse。你可以通过编写代码来配置和使用ClickHouse Sink。以下是一个简单的示例：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.clickhouse.ClickHouseSink;
import org.apache.flink.streaming.connectors.clickhouse.ClickHouseSinkFunction;
import org.apache.flink.streaming.connectors.clickhouse.ClickHouseTableSchema;

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

        // 设置Kafka消费者配置
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test");

        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                "your-topic", new SimpleStringSchema(), properties);

        // 设置ClickHouse生产者配置
        Properties clickHouseProperties = new Properties();
        clickHouseProperties.setProperty("clickhouse.host", "localhost");
        clickHouseProperties.setProperty("clickhouse.port", "9000");
        clickHouseProperties.setProperty("clickhouse.user", "default");
        clickHouseProperties.setProperty("clickhouse.password", "");

        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
                "your-topic", new SimpleStringSchema(), properties);

        // 创建ClickHouse Sink
        ClickHouseTableSchema schema = ClickHouseTableSchema.builder()
                .addColumn("column1", DataType.STRING)
                .addColumn("column2", DataType.INT)
                .build();

        ClickHouseSink<String> clickHouseSink = ClickHouseSink
                .<String>builder()
                .setHost("localhost")
                .setPort(9000)
                .setUser("default")
                .setPassword("")
                .setDatabase("default")
                .setTable("your_table")
                .setSchema(schema)
                .build();

        // 添加数据到ClickHouse Sink
        env.addSource(kafkaConsumer)
                .map(new MapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        return value;
                    }
                })
                .addSink(clickHouseSink);

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

在这个示例中，我们首先创建了一个Flink Kafka消费者和一个Flink Kafka生产者。然后，我们创建了一个ClickHouse Sink，并设置了相关的配置。最后，我们将数据从Kafka消费者传输到ClickHouse Sink。

用vvr-clickhosue连接器，一样的，你想用代码还是sql都可以
缺点。只有fink 1.13 1.15两个版本，此回答整理自钉群“Flink CDC 社区”