Flink CDC里怎么解决？

根据您的描述，您在本地执行Flink CDC时遇到了问题。首先，确保您的本地环境已经正确配置了Flink CDC和Hadoop相关的依赖。然后，您可以尝试以下方法来解决问题：

1. 检查本地的Flink配置文件（如flink-conf.yaml），确保Hadoop相关的配置正确。例如：

hadoop:
  fs.defaultFS: hdfs://localhost:9000
  fs.hdfs.impl: org.apache.hadoop.hdfs.DistributedFileSystem
  fs.checkpointdir: hdfs://localhost:9000/flink/checkpoints

1. 确保您的Flink程序使用了正确的Hadoop连接器。在pom.xml文件中添加以下依赖：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-hadoop_2.11</artifactId>
  <version>1.13.2</version>
</dependency>

1. 在Flink程序中，使用StreamExecutionEnvironment创建流处理任务，并设置CheckpointingMode为EXACTLY_ONCE。例如：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

public class FlinkCDCExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000); // 每5秒进行一次checkpoint
        env.getCheckpointConfig().setCheckpointTimeout(60000); // 设置checkpoint超时时间为60秒
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10))); // 设置重启策略

        // 从Kafka读取数据
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                "input-topic",
                new SimpleStringSchema(),
                PropertiesUtils.fromProperties(new Properties()));
        kafkaConsumer.setStartFromEarliest();
        kafkaConsumer.setGroupId("flink-cdc-group");
        DataStream<String> inputStream = env.addSource(kafkaConsumer);

        // 对数据进行处理
        DataStream<String> processedStream = inputStream
                .map(new MapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        return process(value);
                    }
                })
                .keyBy((value, timestamp) -> value) // 根据值进行分区
                .window(Time.minutes(5)) // 每5分钟一个窗口
                .allowedLateness(Time.seconds(10)); // 允许10秒的延迟

        // 将处理后的数据写入Kafka
        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
                "output-topic",
                new SimpleStringSchema(),
                PropertiesUtils.fromProperties(new Properties()),
                new FlinkFixedPartitioner<>(1));
        processedStream.addSink(kafkaProducer);

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

    private static String process(String value) {
        // 在这里实现您的数据处理逻辑
        return value;
    }
}

1. 如果仍然遇到问题，请检查您的Flink和Hadoop集群的配置，确保它们之间的兼容性。

binlog太多了，一点点过滤，要不你把生产的binlog给清了启动。此回答来自钉群Flink CDC 社区。