Spark Streaming中空batches处理的两种方法（转）-白红宇

Spark Streaming中空batches处理的两种方法（转）

阅读量：6903 次

发布时间：2019-06-27

本文共 1695 字，大约阅读时间需要 5 分钟。

原文链接：

　 Streaming是近实时(near real time)的小批处理系统。对给定的时间间隔(interval)， Streaming生成新的batch并对它进行一些处理。每个batch中的数据都代表一个RDD，但是如果一些batch中没有数据会发生什么事情呢？ Streaming将会产生EmptyRDD的RDD，它的定义如下：

                  01          package org.apache.spark.rdd         
       
                  02                    
       
                  03          import scala.reflect.ClassTag         
       
                  04                    
       
                  05          import org.apache.spark.{Partition, SparkContext, TaskContext}         
       
                  06                    
       
                  07          /**         
       
                  08           * An RDD that has no partitions and no elements.         
       
                  09           */         
       
                  10          private[spark] class EmptyRDD[T: ClassTag](sc: SparkContext) extends RDD[T](sc, Nil) { 
         
                  11                    
       
                  12            override def getPartitions: Array[Partition] = Array.empty         
       
                  13                    
       
                  14            override def compute(split: Partition, context: TaskContext): Iterator[T] = { 
         
                  15              throw new UnsupportedOperationException("empty RDD")         
       
                  16            }         
       
                  17          }

　　可以看到这个RDD并不对任何父RDD有依赖关系，我们不能调用compute方法计算每个分区的数据。EmptyRDD的存在是为了保证Spark Streaming中多个batch的处理是一致的。但是存在EmptyRDD有时候会产生一些问题，比如：如果你想将接收到的Streaming数据写入HDFS中：

                  1          val ssc = new StreamingContext(args(0),"iteblog",Seconds(10))         
       
                  2          val socketStream = ssc.socketTextStream("www.iteblog.com",8888)         
       
                  3          val outputDir = args(1)         
       
                  4                    
       
                  5          socketStream.foreachRDD(rdd => { 
         
                  6            rdd.saveAsTextFile(outputDir)         
       
                  7          })

　　当你调用foreachRDD的时候如果当前rdd是EmptyRDD，这样会导致在HDFS上生成大量的空文件！这肯定不是我们想要的，我们只想在存在数据的时候才写HDFS，我们可以通过以下的两种方法来避免这种情况：

                  1          socketStream.foreachRDD(rdd => { 
         
                  2            if(rdd.count() != 0){ 
         
                  3              rdd.saveAsTextFile(outputDir)         
       
                  4            }         
       
                  5          })

　　EmptyRDD的count肯定是0，所以这样可以避免写空文件，或者我们也可以用下面方法解决：

                  1          socketStream.foreachRDD(rdd => { 
         
                  2            if(!rdd.partitions.isEmpty){ 
         
                  3              rdd.saveAsTextFile(outputDir)         
       
                  4            }         
       
                  5          })

　　EmptyRDD是没有分区的，所以调用partitions.isEmpty是true。这样也可以解决上述问题。

　　虽然上面两种方法都可以解决这个问题，但是推荐使用第二种方法。因为第一种方法调用了RDD的count函数，这是一个Action，会触发一次Job的计算，当你的数据量比较大的时候，这可能会带来性能方面的一些影响；而partitions.isEmpty是不需要触发Job的。

　　不过如果你使用的是Sprk 1.3.0，你可以调用isEmpty函数来判断一个RDD是否为空，这个函数是在SPARK-5270引入的。

转载地址：http://zeldl.baihongyu.com/

你可能感兴趣的文章