执行示例Flink kafka消费者

Question

我正在尝试创建一个简单的Flink Kafka消费者

public class ReadFromKafka {

public static void main(String[] args) throws Exception {
    // create execution environment
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Properties properties = new Properties();
    properties.setProperty("bootstrap.servers", "localhost:9092");
    properties.setProperty("group.id", "flink_consumer");

   DataStream<String> stream = env
            .addSource(new FlinkKafkaConsumer09<>("test", new SimpleStringSchema(), properties));

    stream.map(new MapFunction<String, String>() {
        private static final long serialVersionUID = -6867736771747690202L;

        @Override
        public String map(String value) throws Exception {
            return "Stream Value: " + value;
        }
    }).print();

    env.execute();
}


}

它给了我这个错误：

 INFO  org.apache.kafka.common.utils.AppInfoParser                   - Kafka version: 2.3.0
16:47:28,448 INFO  org.apache.kafka.common.utils.AppInfoParser                   - Kafka commitId: fc1aaa116b661c8a
16:47:28,448 INFO  org.apache.kafka.common.utils.AppInfoParser                   - Kafka startTimeMs: 1563029248441
16:47:28,451 INFO  org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09  - Trying to get partitions for topic test
16:47:28,775 INFO  org.apache.kafka.clients.Metadata                             - [Consumer clientId=consumer-1, groupId=flink_consumer] Cluster ID: 4rz71KZCS_CSasZMrFBNKw
16:47:29,858 INFO  org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09  - Got 1 partitions from these topics: [test]
16:47:29,859 INFO  org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase  - Consumer is going to read the following topics (with number of partitions): 
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/flink/api/java/operators/Keys
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:994)
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.addSource(StreamExecutionEnvironment.java:955)
    at myflink.ReadFromKafka.main(ReadFromKafka.java:43)
Caused by: java.lang.ClassNotFoundException: org.apache.flink.api.java.operators.Keys
    at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:583)
    at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)
    ... 3 more

Process finished with exit code 1

Answer 1

根据您的堆栈跟踪，java找不到类。

Caused by: java.lang.ClassNotFoundException: org.apache.flink.api.java.operators.Keys

这个类在您的依赖项中可能遗漏的flink-java_2.11 jar文件中。

https://www.javadoc.io/doc/org.apache.flink/flink-java_2.11/0.10.2