Flink 接收不到Kafka数据

Question

Kafka集群在虚拟机上，能正常通信能互相ping本地consumer消费者程序正常消费，flink程序不能消费不能打印数据，也没有错误提示

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


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

        // 设置checkpoint
        env.enableCheckpointing(1000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);

        // 创建 KafkaSource
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("hadoop100:9092")
                .setTopics("POC_test")
                .setGroupId("my-groupsdads")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 从 Kafka 读取数据并创建 DataStream
        DataStreamSource<String> weathertest = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "weather_test");

//        System.out.println(weatherTest);
        weathertest.map(value -> {
            System.out.println(value);
            Thread.sleep(10000);
            return value;
        });

        // 打印数据流
//        weathertest.print();

        // 执行任务
        env.execute("CityWeather");
    }
}

Answer 1

从您提供的代码片段来看，有几个问题可能导致您的Flink程序无法正常消费和打印数据：

1. 您在代码中创建了KafkaSource，但没有将其连接到数据流。在创建KafkaSource后，您需要将其作为参数传递给env.fromSource()方法，以便将其连接到数据流。修改您的代码如下：

// 从 Kafka 读取数据并创建 DataStream
DataStreamSource<String> weathertest = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "weather_test");

1. 您的map()操作没有连接到下游操作或输出。在Flink中，转换操作（例如map()）是惰性执行的，只有当您将其连接到下游操作（例如print()）时，它才会被执行。您可以在map()操作后添加.print()方法，以打印数据流中的结果，或者将map()操作的结果连接到其他下游操作。修改您的代码如下：

weathertest.map(value -> {
    System.out.println(value);
    Thread.sleep(10000);
    return value;
}).print(); // 将结果打印出来

1. 您没有调用execute()方法来触发任务执行。在您的代码末尾，添加env.execute("CityWeather");以启动任务执行。修改您的代码如下：

// 执行任务
env.execute("CityWeather");

请确保在代码中添加了上述修改，然后重新运行您的程序，看看是否能够正常消费和打印数据。