foreachRDD不执行
foreachRDD是Apache Spark Streaming中的一个函数，用于对DStream中的每个RDD执行指定的操作。但是如果foreachRDD不执行，可能有以下几个原因：
- DStream没有被启动：在Spark Streaming中，DStream需要通过调用start()方法来启动。如果没有启动DStream，那么其中的操作，包括foreachRDD，将不会执行。可以通过调用streamingContext.start()来启动DStream。
- 没有设置输出操作：foreachRDD需要在DStream中设置输出操作，才能执行。输出操作可以是将数据保存到文件系统、数据库或其他外部系统中。如果没有设置输出操作，foreachRDD将不会执行。可以通过调用DStream的saveAsTextFiles()、foreach()等方法来设置输出操作。
- 没有触发计算：Spark Streaming是基于微批处理的，需要通过调用streamingContext.awaitTermination()或streamingContext.stop()来触发计算。如果没有触发计算，foreachRDD将不会执行。可以在代码的最后添加streamingContext.awaitTermination()来触发计算。
总结起来，如果foreachRDD不执行，需要确保DStream已经启动、设置了输出操作，并且触发了计算。如果仍然无法执行，可能需要检查代码中是否存在其他问题，例如数据源是否正确、数据处理逻辑是否正确等。
腾讯云相关产品和产品介绍链接地址：
- 腾讯云云服务器（CVM）：https://cloud.tencent.com/product/cvm
- 腾讯云云数据库 MySQL 版（CDB）：https://cloud.tencent.com/product/cdb
- 腾讯云云原生容器服务（TKE）：https://cloud.tencent.com/product/tke
- 腾讯云人工智能（AI）：https://cloud.tencent.com/product/ai
- 腾讯云物联网（IoT）：https://cloud.tencent.com/product/iot
- 腾讯云移动开发（移动推送）：https://cloud.tencent.com/product/umeng
- 腾讯云对象存储（COS）：https://cloud.tencent.com/product/cos
- 腾讯云区块链服务（BCS）：https://cloud.tencent.com/product/bcs
- 腾讯云游戏多媒体引擎（GME）：https://cloud.tencent.com/product/gme
- 腾讯云音视频处理（VOD）：https://cloud.tencent.com/product/vod
- 腾讯云网络安全（DDoS 高防 IP）：https://cloud.tencent.com/product/ddos
相关·内容
1回答
为什么foreachRDD没有执行？这是代码，不是在stream.foreachRDD执行开始和在}结束的部分)；。感谢您的回答;)。); stream.foreachRDD
浏览 17提问于2018-08-21得票数 2
这是我简化的Apache Spark Streaming代码，它通过Kafka Streams获得输入，组合，打印并保存到一个文件中。但是现在我希望传入的数据流保存在MongoDB中。 .setAppName("StreamingDataToMongoDB")
.set("spark.streaming.concurrentJobs", "
浏览 0提问于2016-12-18得票数 0
我们有一个星火流应用程序，它摄取数据@10,000/秒.我们在我们的foreachRDD上使用DStream操作(因为spark不执行，除非它在DStream上找到输出操作)CodeSnippet 1:
rdd.foreach，所以注释掉了它，看看花了多少时间.让我感到惊讶的是，在f
浏览 11提问于2017-03-03得票数 1
根据我读过的和文档，星火流的foreachRDD( someFunction )只会在驱动程序进程中执行someFunction本身，尽管如果对RDD进行操作，那么它们将在执行程序上执行-- RDDs位于的位置尽管我注意到，如果打开检查点，那么spark似乎正在尝试序列化foreachRDD(someFunction)中的所有内容并发送到某个地方--这对我来说也同样有效，这给我带来了麻烦，因为使用的对象之一是不可序列化的是否有一种方法可以让Spark不序列化foreachRDD<
浏览 9提问于2016-09-22得票数 3
回答已采纳
stream.foreachRDD( new Function<JavaRDD<String>,JavaRDD<String> >() {
// do some stuffreturn RDD下面是我正在收到的错误：
方法foreachRDD(Function
浏览 2提问于2017-08-01得票数 1
回答已采纳
2回答
在这种情况下dStream.foreachRDD(a => ... )执行以下方法：foreachRDD( a => )在foreachRDD( b => )之前？
val dStream : Stream[_] = ...().cache()
浏览 1提问于2016-05-24得票数 4
回答已采纳
KafkaUtils.createDirectStream[...](...)mappedStream.foreachRDD{ ... }mappedStream.map { ... }.foreachRDD { ... }
是否有一种方法可以注册最后一个foreachRDD，保证在上面的foreachRDD完成执行</
浏览 11提问于2017-02-01得票数 5
回答已采纳
我想在DStream上执行一系列操作。动作N+1必须在动作N之后执行，这些实现之间有什么区别？ val myDStream = ???//version 1myDStream.foreachRDD(rdd => action 2)myDStream.foreac
浏览 20提问于2019-01-23得票数 0
回答已采纳
1回答
(StreamingContext.scala:572)，org.apache.spark.streaming，scala.Predef$.require(Predef.scala:224)没有什么可执行的(StreamingContext.scala:572)，org.apache.spark.streaming，scala.Predef$.require(Predef.scala:224)没有什么可执行的myRDD2.foreachRDD(itr -> new VoidFunction<Iterator<String&
浏览 4提问于2017-04-11得票数 0
回答已采纳
据我所知，我可以为底层数据创建一个流，然后进入foreachRDD来运行我的计算。但这假设我知道所有的计算，我需要做一个‘先验。当新用户需要新的计算时，是否可以通过在单独的线程中进入foreachRDD来添加额外的计算？
提前谢谢。
浏览 1提问于2015-10-02得票数 1
在spark streaming应用程序中，如何在lines.foreachRDD()块完成执行后执行lines.map()函数。DataTuple> initFunction(JavaDStream<DataTuple> lines, final String str) throws Exception {
return lines;我
浏览 3提问于2016-05-24得票数 0
1回答
instance = jsc.broadcast(wordBlacklist); } return instance;}
并在wordCounts.foreachRDD我的问题是，为什么不直接在父类中声明private static volatile Broadcast<List<String>> instance = null;，即JavaRecoverableNetworkWordCount(我认为，由于广播变量是在foreachRDD()中初始化的，在单个驱动程序线程中
浏览 0提问于2018-01-04得票数 0
回答已采纳
我从kafka获取的流数据是hdfs文件的路径，我需要获取该文件的数据。 public String call(Tuple2<String, String> arg0) StringReader reader = new St
浏览 19提问于2017-01-10得票数 0
如何在foreachRDD块中进行微批处理操作。例如，我从HDFS读取日志并在foreachRDD中执行操作lines.foreachRDD{ rdd =>
浏览 2提问于2016-03-16得票数 0
2回答
val summary = new ArrayBuffer[(String,Int)]() summary
浏览 4提问于2015-02-25得票数 4
回答已采纳
1回答
背景
我有一个火花流应用程序，它从->读取数据，对其进行窗口化，->将数据保存到外部系统(通过执行foreachRDD)。最近我观察到，我的窗户是由foreachRDD一个接一个地消耗的.这意味着如果我的应用程序中有突然的数据突发(因此窗口的foreachRDD需要很长时间)，那么在处理之前，这些窗口将堆在队列中(而我的集群中的大多数机器都是空闲的如果是，是否有任何方法可以并行地进行“窗口”操作，从而使foreachRDD同时使用窗口？
浏览 3提问于2016-03-02得票数 0
1回答
= ssc.sparkContext.longAccumulator("numInst")input.foreachRDD
浏览 1提问于2018-05-12得票数 0
回答已采纳
对于下面的代码，.count()是将值返回给驱动程序还是仅返回给执行器？ long count = rdd.count();
// some code to save count to Datastore对于将来的其他相关问题，有没有一种简单的方法来验证代码块是否在驱动程序或执行器上执行？
浏览 5提问于2017-07-11得票数 2
回答已采纳
但是，根据foreachRdd方法在DStream上的API文档:它在驱动程序上执行，这意味着所有处理逻辑只能在驱动程序上运行，而不是分发给工人/执行者。API文档
将函数func应用于从流生成的每个RDD的最通用的输出操作符。请注意，函数func是在运行流应用程序的驱动进程中执行的，其中通常包含RDD操作，这将强制计算流RDD。
浏览 1提问于2016-07-22得票数 13
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
