对DStream.foreachRDD的理解

Posted 2022-07-30 代立冬

 
  
最近在使用Spark Streaming过程中，对foreachRDD有点疑问，查阅资料后记录如下：
  


  
foreachRDD(func)的官方解释为

  

   
The most generic output operator that applies a function, func, to each RDD generated from the stream. This function should push the data in each RDD to an external system, such as saving the RDD to files, or writing it over the network to a database. Note that the function func is executed in the driver process running the streaming application, and will usually have RDD actions in it that will force the computation of the streaming RDDs.
  

  
对于这个定义会产生一个疑问：在一个batch interval里面会产生几个RDD？
结论：有且只有一个。

  
那么定义里面所说的“each RDD”应该如何理解呢？

  
DStream可以理解为是基于时间的，即每个interval产生一个RDD，所以如果以时间为轴，每隔一段时间就会产生一个RDD，那么定义中的“each RDD”应该理解为每个interval的RDD，而不是一个interval中的每个RDD。

  
从spark的源码分析

  
DStream中的foreachRDD方法最终会调用如下的代码

  
private def foreachRDD(
    foreachFunc: (RDD[T], Time) => Unit,
    displayInnerRDDOps: Boolean): Unit = 
  new ForEachDStream(this,
    context.sparkContext.clean(foreachFunc, false), displayInnerRDDOps).register()

  
可以看到这个方法里面并没有任何的Iterator，可以对比一下RDD中的foreachPartition和foreach方法，这两个方法是会遍历RDD，所以才会有Iterator类型的引用
  
def foreach(f: T => Unit): Unit = withScope 
  val cleanF = sc.clean(f)
  sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))


def foreachPartition(f: Iterator[T] => Unit): Unit = withScope 
  val cleanF = sc.clean(f)
  sc.runJob(this, (iter: Iterator[T]) => cleanF(iter))

  
而如果每个interval中有多个RDD，那么DStream中的foreachRDD也一定会有Iterator类型的引用，但是从上述的代码中并没有。
 

作者：Woople
链接：http://www.jianshu.com/p/9116043b0c21