原创力文档- 1、本文档共2页，可阅读全部内容。
- 2、原创力文档（book118）网站文档一经付费（服务费），不意味着购买了该文档的版权，仅供个人/单位学习、研究之用，不得用于商业用途，未经授权，严禁复制、发行、汇编、翻译或者网络传播等，侵权必究。
- 3、本站所有内容均由合作方或网友上传，本站不对文档的完整性、权威性及其观点立场正确性做任何保证或承诺！文档内容仅供研究参考，付费前请自行鉴别。如您付费，意味着您自己接受本站规则且自行承担风险，本站不退款、不进行额外附加服务；查看《如何避免下载的几个坑》。如果您已付费下载过本站文档，您可以点击 这里二次下载。
- 4、如文档侵犯商业秘密、侵犯著作权、侵犯人身权等，请点击“版权申诉”（推荐），也可以打举报电话：400-050-0827(电话支持时间：9:00-18:30)。
查看更多
SparkStreaming中空batches处理的两种方法.PDF
Spark Streaming中空batches处理的两种方法
Spark大数据博客 -
Spark Streaming中空batches处理的两种方法
Spark Streaming是近实时(near real
time)的小批处理系统。对给定的时间间隔(interval)，Spark Streaming生成新的batch并对它进行
一些处理。每个batch中的数据都代表一个RDD，但是如果一些batch中没有数据会发生什么事情
呢？Spark Streaming将会产生EmptyRDD的RDD，它的定义如下：
package org.apache.spark.rdd
import scala.reflect.ClassTag
import org.apache.spark.{Partition, SparkContext, TaskContext}
/**
* An RDD that has no partitions and no elements.
*/
private[spark] class EmptyRDD[T: ClassTag](sc: SparkContext) extends RDD[T](sc, Nil) {
override def getPartitions: Array[Partition] = Array.empty
override def compute(split: Partition, context: TaskContext): Iterator[T] = {
throw new UnsupportedOperationException(empty RDD)
}
}
可以看到这个RDD并不对任何父RDD有依赖关系，我们不能调用compute方法计算每个分区
的数据。EmptyRDD的存在是为了保证Spark Streaming中多个batch的处理是一致的。但是存在E
mptyRDD有时候会产生一些问题，比如：如果你想将接收到的Streaming数据写入HDFS中：
val ssc = new StreamingContext(args(0),iteblog,Seconds(10))
val socketStream = ssc.socketTextStream(,8888)
val outputDir = args(1)
socketStream.foreachRDD(rdd = {
rdd.saveAsTextFile(outputDir)
})
1 / 2
Spark Streaming中空batches处理的两种方法
Spark大数据博客 -
当你调用foreachRDD的时候如果当前rdd是EmptyRDD，这样会导致在HDFS上生成大量的空
文件！这肯定不是我们想要的，我们只想在存在数据的时候才写HDFS，我们可以通过以下的两种
方法来避免这种情况：
socketStream.foreachRDD(rdd = {
if(rdd.count() != 0){
rdd.saveAsTextFile(outputDir)
}
})
EmptyRDD的count肯定是0，所以这样可以避免写空文件，或者我们也可以用下面方法解决
：
socketStream.foreachRDD(rdd = {
if(!rdd.partitions.isEmpty){
rdd.saveAsTextFile(outputDir)
}
})
EmptyRDD是没有分区的，所以调用partitions.isEmpty是true。这样也可以
文档评论（0）