如何将Spark和Kafka集成到直接流中

作者: 早岁那知世事艰
发布时间: 2024-12-13 08:40:14 (13天前)
转自：

5 条回复

0#
回复此人
老夫的少女心 | 2019-08-31 10-32

<div class =“post-text”itemprop =“text”> <P> 我认为你没有使用ssc.start（）启动流 </p> </DIV>

编辑
1#
回复此人
晴天？霹雳 | 2019-08-31 10-32

<div class =“post-text”itemprop =“text”> <P> 为您的代码添加一个动作可能有所帮助 </p> <pre> <code> stream.print() </code> </pre> <P> 试一试，祝你好运 </p> </DIV>

编辑
2#
回复此人
晴天3 | 2019-08-31 10-32

<div class =“post-text”itemprop =“text”> <P> 我认为日志说明你需要的一切:) </p> <BLOCKQUOTE> <P> IllegalArgumentException：要求失败：未注册任何输出操作，因此无需执行任何操作 </p> </BLOCKQUOTE> <P> 什么是输出操作？例如： </p> <UL> <LI> foreachRDD </LI> <LI> 打印 </LI> <LI> saveAsHadoopFile </LI> <LI> 和别的。你可以得到更多 <a href="http://spark.apache.org/docs/latest/streaming-programming-guide.html#output-operations-on-dstreams" rel="nofollow noreferrer"> 链接 </A> 到文档。 </LI> </UL> <P> 您必须向应用程序添加一些操作，例如save <code> stream.mapToPair </code> 变量然后在这个变量或print（）上调用foreachRDD来显示值 </p> </DIV>

编辑
3#
回复此人
Jacob | 2019-08-31 10-32

<div class =“post-text”itemprop =“text”> <P> 使用Kafka创建直接流后，您将创建 <code> JavaPairDStream </code> 。现在你可以迭代了 <code> JavaPairDStream </code> 并打印您的Kafka消息的密钥和值。 </p> <pre> <code> JavaPairDStream<String, String> jPairDStream = stream.mapToPair( new PairFunction<ConsumerRecord<String, String>, String, String>() { @Override public Tuple2<String, String> call(ConsumerRecord<String, String> record) throws Exception { return new Tuple2<>(record.key(), record.value()); } }); jPairDStream.foreachRDD(jPairRDD -> { jPairRDD.foreach(rdd -> { System.out.println("key="+rdd._1()+" value="+rdd._2()); }); }); jssc.start(); jssc.awaitTermination(); </code> </pre> <P> 您正在使用的命令 <code> kafka-console-producer.bat </code> 将产生一个消息，其中键将为null。为了生成主题上具有键和值的消息 <code> test </code> 使用以下命令。在控制台中输入逗号分隔的键和值 <code> key1,value1 </code> </p> <pre> <code> kafka-console-producer.bat --property parse.key=true --property key.separator=, --broker-list localhost:9092 --topic test </code> </pre> <P> 您的pom文件具有不同的spark构件版本。确保对所有工件使用相同的版本。您需要以下依赖项来运行您的程序。 </p> <pre> <code> <dependencies> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>2.1.0</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>2.1.0</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_2.11</artifactId> <version>2.1.0</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-kafka-0-10_2.11</artifactId> <version>2.1.0</version> </dependency> </dependencies> </code> </pre> </DIV>

编辑

登录后才能参与评论