Spark算子执行流程详解之二

Posted 亮亮-AC米兰

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark算子执行流程详解之二相关的知识,希望对你有一定的参考价值。

4.count

def count(): Long = sc.runJob(this, Utils.getIteratorSize_).sum

计算数据总量,每个分区各自计算自己的总数,然后汇总到driver端,driver端再把每个分区的总数相加统计出对应rdd的数据量,其流程如下:

 


5.countApprox

在一定的超时时间之内返回rdd元素的个数,其rdd元素的总数分布符合正态分布,其分布因子为confidence,当超过timeout时,返回一个未完成的结果。

/**
 * :: Experimental ::
 * Approximate version of count() that returns a potentially incomplete result
 * within a timeout, even if not all tasks have finished.
 */
@Experimental
def countApprox(
    timeout: Long,
    confidence: Double = 0.95): PartialResult[BoundedDouble] = withScope

//定义在excutor端计算总数的函数
  val
countElements: (TaskContext, Iterator[T]) => Long = (ctx, iter) =>
    var result = 0L
   
while (iter.hasNext)
      result += 1L
     
iter.next()
   
    result
 

//定义在driver端的一个监听回调函数,当task完成的时候,会触发里面的merge操作,当超时时间到之后或者任务提前完成的话,会取里面的当前状态,即currentResult
  val
evaluator = newCountEvaluator(partitions.length, confidence)

//提交任务
  sc.runApproximateJob(this, countElements, evaluator, timeout)

继续往下看,看看evaluator是如何执行的:

def runApproximateJob[T,U, R](
    rdd: RDD[T],
    func: (TaskContext, Iterator[T]) =>U,
    evaluator: ApproximateEvaluator[U, R],
    timeout: Long): PartialResult[R] =
  assertNotStopped()
  val callSite = getCallSite
  logInfo("Starting job: " + callSite.shortForm)
  val start = System.nanoTime
 
val cleanedFunc = clean(func)

// cleanedFunc就是countElementsevaluator就是CountEvaluator,超时时间为timeout
  val
result = dagScheduler.runApproximateJob(rdd, cleanedFunc, evaluator, callSite, timeout,
    localProperties.get)
  logInfo(
    "Job finished: " + callSite.shortForm +", took " + (System.nanoTime- start) / 1e9 + " s")
  result

继续看runApproximateJob的实现:

def runApproximateJob[T,U, R](
    rdd: RDD[T],
    func: (TaskContext, Iterator[T]) =>U,
    evaluator: ApproximateEvaluator[U, R],
    callSite: CallSite,
    timeout: Long,
    properties: Properties): PartialResult[R] =

//定义一个监听器,当有任务完成的时候触发taskSucceeded,当超时时间到的时候返回CountEvaluator的当前值
  val
listener = newApproximateActionListener(rdd, func, evaluator, timeout)
  val func2 = func.asInstanceOf[(TaskContext,Iterator[_]) => _]
  val partitions = (0until rdd.partitions.size).toArray
  val jobId = nextJobId.getAndIncrement()

//提交任务
  eventProcessLoop.post(JobSubmitted(
    jobId, rdd, func2, partitions, allowLocal = false, callSite, listener,
    SerializationUtils.clone(properties)))

//等待计算结果
  listener.awaitResult()    // Will throw an exception if the job fails

因此其超时计算总数的逻辑主要在ApproximateActionListener里面,请看ApproximateActionListener:

private[spark] classApproximateActionListener[T, U, R](
    rdd: RDD[T],
    func: (TaskContext, Iterator[T]) =>U,
    evaluator: ApproximateEvaluator[U, R],
    timeout: Long)
  extends JobListener

  val startTime= System.currentTimeMillis()
  val totalTasks= rdd.partitions.size
  var finishedTasks= 0
 
var failure: Option[Exception] = None             // Set if the job has failed (permanently)
 
var resultObject: Option[PartialResult[R]] = None// Set if we've already returned a PartialResult
 

//当某个分区完成的时候触发taskSucceeded回调函数
 
override def taskSucceeded(index: Int, result: Any)
    synchronized

//更新CountEvaluator的当前值
      evaluator.merge(index, result.asInstanceOf[U])
      finishedTasks += 1
     
if (finishedTasks== totalTasks) //当全部分区都完成的是退出等待,返回计算结果
        // If we had already returned a PartialResult, set its final value
       
resultObject
.foreach(r => r.setFinalValue(evaluator.currentResult()))
        // Notify any waiting thread that may have called awaitResult

//退出等待
       
this.notifyAll()
     
   
 
  ……
  /**
   * Waits for up to timeout milliseconds since the listener was created and then returns a
   * PartialResult with the result so far. This may be complete if the whole job is done.
   */

//等待计算结果
 
def awaitResult(): PartialResult[R] = synchronized
    val finishTime = startTime+ timeout
    while (true)
      val time = System.currentTimeMillis()
      if (failure.isDefined)
        throw failure.get
      else if (finishedTasks== totalTasks) //如果在超时时间之内计算完成,则返回全部结果
        return new
PartialResult(evaluator.currentResult(),true)
      else if (time >= finishTime) //如果已经超时,则返回部分结果
        resultObject = Some(newPartialResult(evaluator.currentResult(), false))
        return resultObject.get
      else //如果超时时间没到,则继续休眠
        this
.wait(finishTime - time)
     
   
    // Should never be reached, but required to keep the compiler happy
   
return null
 

其中如果在超时时间之内没有完成的话,evaluator.currentResult()会返回符合总数符合正态分布的一个近似结果,感兴趣的同学可以继续研究下去:

private[spark] classCountEvaluator(totalOutputs: Int, confidence: Double)
  extends ApproximateEvaluator[Long, BoundedDouble]

  var outputsMerged= 0
 
var sum: Long =0

 
override def merge(outputId: Int, taskResult: Long)
    outputsMerged += 1
   
sum += taskResult
 

  override def currentResult(): BoundedDouble =
    if (outputsMerged== totalOutputs) //全部完成
      new
BoundedDouble(sum,1.0, sum,sum)
    else if (outputsMerged== 0) //一个任务都没完成
      new
BoundedDouble(0,0.0, Double.NegativeInfinity, Double.PositiveInfinity)
    else //部分完成,计算其理论总数的正态分布参数
      val
p = outputsMerged.toDouble / totalOutputs
      val mean = (sum+ 1 - p) / p
      val variance = (sum+ 1) * (1 - p) / (p * p)
      val stdev = math.sqrt(variance)
      val confFactor = newNormalDistribution().
        inverseCumulativeProbability(1 - (1- confidence) / 2)
      val low = mean - confFactor * stdev
      val high = mean + confFactor * stdev
      new BoundedDouble(mean, confidence, low, high)
   
 

因此countApprox的计算过程大致如下:1)excutor端不断的计算分区的总数然后上报给driver端;2)driver端接受excutor上报的总数进行统计,如果在超时时间之内没有全部上报完成的话,则强制退出,返回一个其总数符合正态分布的值,如果在超时时间之内计算完成的话,则返回一个准确值。


6.countApproxDistinct

作用是对RDD集合内容进行去重统计,该统计是一个大约的统计,参数relativeSD控制统计的精确度。relativeSD越小,结果越准确。

/**
 * Return approximate number of distinct elements in the RDD.
 *
 * The algorithm used is based on streamlib's implementation of "HyperLogLog in Practice:
 * Algorithmic Engineering of a State of The Art Cardinality Estimation Algorithm", available
 *
<a href="http://dx.doi.org/10.1145/2452376.2452456">here</a>.
 *
 *
@param relativeSD Relative accuracy. Smaller values create counters that require more space.
 *                   It must be greater than 0.000017.
 */
def countApproxDistinct(relativeSD: Double =0.05): Long = withScope
  require(relativeSD > 0.000017, s"accuracy ($relativeSD) must be greater than 0.000017")
  val p = math.ceil(2.0* math.log(1.054 / relativeSD) / math.log(2)).toInt
  countApproxDistinct(if (p < 4) 4 elsep, 0)

采用的是HyperLogLog in Practice算法,原理比较深奥,有兴趣的可以深究。
实例如下: 

object CountApproxDistinct

  def main(args: Array[String])

    val conf = new SparkConf().setAppName("spark-demo").setMaster("local")

    val sc = new SparkContext(conf)

    /**

     * 构建一个集合,分成20个partition

     */

    val a = sc.parallelize(1 to 10000 , 20)

    //RDD a内容复制5遍,其中有50000个元素

    val b = a++a++a++a++a

    //结果是9760,不传参数,默认是0.05

    println(b.countApproxDistinct())

    //结果是9760

    println(b.countApproxDistinct(0.05))

    //8224

    println(b.countApproxDistinct(0.1))

    //10000

    println(b.countApproxDistinct(0.001))

 

 

7.collect

 

def collect(): Array[T] = withScope
  val results = sc.runJob(this, (iter:Iterator[T]) => iter.toArray)
  Array.concat(results: _*)

获取Rdd的所有数据,然后缓存在Driver端,其流程如下:


如果RDD数据量很大的话,请谨慎使用,因为会缓存该RDD的所有数据量。


8.toLocalIterator

返回一个保护所有记录的迭代器

/**
 * Return an iterator that contains all of the elements in this RDD.
 *
 * The iterator will consume as much memory as the largest partition in this RDD.
 *
 * Note: this results in multiple Spark jobs, and if the input RDD is the result
 * of a wide transformation (e.g. join with different partitioners), to avoid
 * recomputing the input RDD should be cached first.
 */
def toLocalIterator:Iterator[T] = withScope

//针对每个分区触发一次action
  def
collectPartition(p: Int): Array[T] =
    sc.runJob(this, (iter: Iterator[T]) => iter.toArray, Seq(p), allowLocal =false).head
 

//调用flatMap将所有记录组装起来返回单个迭代器
  (0 until partitions.length).iterator.flatMap(i => collectPartition(i))

即:

scala> val rdd = sc.parallelize(1 to 10,2)

rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> val it = rdd.toLocalIterator

it: Iterator[Int] = non-empty iterator

scala> while(it.hasNext)

     | println(it.next)

     |

1

2

3

4

5

6

7

8

9

10


9.takeOrdered

takeOrdered函数用于从RDD中,按照默认(升序)或指定排序规则,返回前num个元素。

def takeOrdered(num: Int)(implicitord: Ordering[T]): Array[T] = withScope
  if (num == 0)
    Array.empty
 
else
    val mapRDDs = mapPartitions items =>

//先在excutor端进行排序,按照ord排序规则,转化为前num个优先队列
      // Priority keeps the largest elements, so let's reverse the ordering.
     
val queue = new BoundedPriorityQueue[T](num)(ord.reverse)
      queue ++= util.collection.Utils.takeOrdered(items, num)(ord)
      Iterator.single(queue)
   
    if (mapRDDs.partitions.length ==0)
      Array.empty
   
else

//将分区的计算结果传送给driver,转化为数组,进行排序取前num条记录
      mapRDDs.reduce (queue1, queue2) =>
        queue1 ++= queue2
        queue1
      .toArray.sorted(ord)
   
 

例如:

List<Integer> data = Arrays.asList(1,4,3,2,5,6);
JavaRDD<Integer> JavaRDD = jsc.parallelize(data,2);
for(Integer integer:JavaRDD.takeOrdered(2))
    System.out.println(integer);

打印

1

2

其执行流程如下:

 

 



以上是关于Spark算子执行流程详解之二的主要内容,如果未能解决你的问题,请参考以下文章

Spark算子执行流程详解之八

Spark算子执行流程详解之一

Spark算子执行流程详解之五

Spark算子执行流程详解之四

Spark算子执行流程详解之七

Spark算子执行流程详解之七