java - 执行示例 Flink kafka 消费者
问题描述
我正在尝试创建一个简单的 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
解决方案
根据您的堆栈跟踪,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
推荐阅读
- firebase - Flutter如何对集合中的文档进行排序,其中包含每个文档的编号字段
- html - 具有自适应 img 高度的等高项的 Flex 列
- web-services - 与 Azure 支持一起使用时,Xamarin 主从模板是否损坏?
- flutter - 在 Flutter 中通过主题设置按钮的颜色
- python - 弧形/弯曲文本图像的最佳方法是什么?
- javascript - 如何隐藏复选框的“框”并只保留单词标签
- python - 如何在 Django 中使用 sympy 构建包含 python 表达式的文本区域的 LaTeX 实时预览?
- vue.js - 捆绑组件的 currentinstance 值为 null
- android - Firestore 黑客案例研究
- c - double 表示最多 16 位有效数字的值