从kafka控制台生成器将数据读入flink程序

xhv8bpkk  于 2021-06-08  发布在  Kafka
关注(0)|答案(1)|浏览(382)

我创建了一个名为 test 并使用控制台生成器在控制台中编写了一些字符串。

  1. ./bin/kafka-console-producer.sh --topic test --broker-list localhost:9092

幸运的是，我能够通过使用 console-consumer . 现在，我想使用 console-consumer 在flink程序中使用下面的代码

  1. public class ReadFromKafka {
  3.     public static void main(String[] args) throws Exception {
  4.         // create execution environment
  5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  7.         Properties properties = new Properties();
  8.         properties.setProperty("bootstrap.servers", "localhost:9092");
  9.         properties.setProperty("zookeeper.connect", "localhost:2181");
  10.         properties.setProperty("group.id", "test");
  12.         DataStream<String> message = env.addSource(new FlinkKafkaConsumer08<String>("test", new SimpleStringSchema(),properties));
  14.         message.map(new MapFunction<String, String>() {
  15.             private static final long serialVersionUID = -6867736771747690202L;
  17.             @Override
  18.             public String map(String value) throws Exception {
  19.                 return " Value: " + value;
  20.             }
  21.         }).print();
  23.         env.execute();
  25.     } //main
  26. } //ReadFromKafka

pom.xml的内容如下

  1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  2.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  6.     <groupId>org.stsffap</groupId>
  7.     <artifactId>cep-monitoring</artifactId>
  8.     <name>cep-monitoring</name>
  9.     <version>1.0</version>
  11.     <packaging>jar</packaging>
  13.     <properties>
  14.         <flink.version>1.0.1</flink.version>
  16.         <slf4j.version>1.7.7</slf4j.version>
  17.         <log4j.version>1.2.17</log4j.version>
  18.     </properties>
  20.     <dependencies>
  22.         <dependency>
  23.             <groupId>org.slf4j</groupId>
  24.             <artifactId>slf4j-log4j12</artifactId>
  25.             <version>${slf4j.version}</version>
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>log4j</groupId>
  29.             <artifactId>log4j</artifactId>
  30.             <version>${log4j.version}</version>
  31.         </dependency>
  33.         <dependency>
  34.             <groupId>org.apache.flink</groupId>
  35.             <artifactId>flink-streaming-java_2.10</artifactId>
  36.             <version>${flink.version}</version>
  37.         </dependency>
  38. <dependency>
  39.         <groupId>org.apache.flink</groupId>
  40.         <artifactId>flink-java</artifactId>
  41.         <version>1.3.2</version>
  42.     </dependency>
  44.         <dependency>
  45.     <groupId>org.apache.flink</groupId>
  46.     <artifactId>flink-clients_2.10</artifactId>
  47.     <version>1.3.2</version>
  48.     </dependency>
  50. <dependency>
  51.             <groupId>org.apache.flink</groupId>
  52.             <artifactId>flink-cep_2.10</artifactId>
  53.             <version>${flink.version}</version>
  54.         </dependency>
  56.         <dependency>
  57.             <groupId>org.apache.flink</groupId>
  58.             <artifactId>flink-connector-kafka-0.8_2.10</artifactId>
  59.             <version>1.3.2</version>
  61.         </dependency>
  63.     </dependencies>
  65.     <build>
  66.         <plugins>
  68.             <plugin>
  69.                 <groupId>org.apache.maven.plugins</groupId>
  70.                 <artifactId>maven-compiler-plugin</artifactId>
  71.                 <version>3.1</version>
  73.                 <configuration>
  74.                     <source>1.8</source>
  75.                     <target>1.8</target>
  76.                     <compilerId>jdt</compilerId>
  77.                 </configuration>
  78.                 <dependencies>
  79.                     <dependency>
  80.                         <groupId>org.eclipse.tycho</groupId>
  81.                         <artifactId>tycho-compiler-jdt</artifactId>
  82.                         <version>0.21.0</version>
  83.                     </dependency>
  84.                 </dependencies>
  85.             </plugin>
  87.         </plugins>
  88.     </build>
  89. </project>

每当我执行这个代码时，我都会遇到以下错误

  1. objc[892]: Class JavaLaunchHelper is implemented in both /Library/Java/JavaVirtualMachines/jdk1.8.0_144.jdk/Contents/Home/bin/java (0x109f654c0) and /Library/Java/JavaVirtualMachines/jdk1.8.0_144.jdk/Contents/Home/jre/lib/libinstrument.dylib (0x10afd44e0). One of the two will be used. Which one is undefined.
  2. Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/flink/streaming/api/checkpoint/CheckpointedFunction
  3.     at java.lang.ClassLoader.defineClass1(Native Method)
  4.     at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
  5.     at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
  6.     at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
  7.     at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
  8.     at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
  9.     at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
  10.     at java.security.AccessController.doPrivileged(Native Method)
  11.     at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
  12.     at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
  13.     at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
  14.     at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
  15.     at java.lang.ClassLoader.defineClass1(Native Method)
  16.     at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
  17.     at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
  18.     at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
  19.     at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
  20.     at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
  21.     at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
  22.     at java.security.AccessController.doPrivileged(Native Method)
  23.     at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
  24.     at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
  25.     at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
  26.     at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
  27.     at org.stsffap.cep.monitoring.ReadFromKafka.main(ReadFromKafka.java:24)
  28. Caused by: java.lang.ClassNotFoundException: org.apache.flink.streaming.api.checkpoint.CheckpointedFunction
  29.     at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
  30.     at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
  31.     at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
  32.     at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
  33.     ... 25 more

也是Kafka的一个版本，我使用下面的find by命令

  1. find ./libs/ -name \*kafka_\* | head -1 | grep -o '\kafka[^\n]*'

是

  1. kafka_2.11-0.9.0.0-javadoc.jar

我是否需要使用.8.x版本的Kafka来运行我的示例？

对意见和建议高度赞赏。提前谢谢。祝你好运！

Mavenapache-kafkaapache-flinkflink-streamingapache-kafka-connect

1条答案

按热度按时间
gopyfrb3

gopyfrb31#

我的程序开始工作时做了以下修改，我把Kafka的版本更新到了.9.x

  1. <dependency>
  2.             <groupId>org.apache.flink</groupId>
  3.             <artifactId>flink-connector-kafka-0.9_2.10</artifactId>
  4.             <version>${flink.version}</version>
  6.         </dependency>

我将flink版本从1.0.1升级到1.1.2，如下所示

  1. <properties>
  2.         <!-- <flink.version>1.0.1</flink.version>-->
  4.         <flink.version>1.1.2</flink.version>
  5.         <slf4j.version>1.7.7</slf4j.version>
  6.         <log4j.version>1.2.17</log4j.version>
  7.     </properties>
  9.     <dependencies>
  11.         <dependency>
  12.             <groupId>org.slf4j</groupId>
  13.             <artifactId>slf4j-log4j12</artifactId>
  14.             <version>${slf4j.version}</version>
  15.         </dependency>
  16.         <dependency>
  17.             <groupId>log4j</groupId>
  18.             <artifactId>log4j</artifactId>
  19.             <version>${log4j.version}</version>
  20.         </dependency>
  22.         <dependency>
  23.             <groupId>org.apache.flink</groupId>
  24.             <artifactId>flink-streaming-java_2.10</artifactId>
  25.             <version>${flink.version}</version>
  26.         </dependency>
  27. <dependency>
  28.         <groupId>org.apache.flink</groupId>
  29.         <artifactId>flink-java</artifactId>
  30.         <version>${flink.version}</version>
  31.     </dependency>
  33.         <dependency>
  35.     <groupId>org.apache.flink</groupId>
  36.     <artifactId>flink-clients_2.10</artifactId>
  37.     <version>${flink.version}</version>
  38.     </dependency>
展开查看全部
赞(0）分享  回复(0）举报  2021-06-08
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com