spark任务中基于rdd的执行流程分析

57 阅读 0 评论 38 点赞

我是靠谱客的博主现实烧鹅，最近开发中收集的这篇文章主要介绍spark任务中基于rdd的执行流程分析，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

下面是rdd的论文中，从hdfs读取日志数据的一个例子：

 
lines = spark.textFile("hdfs://...")  // lines is a org.apache.spark.rdd.MappedRDD

errors = lines.filter(_.startsWith("ERROR")) // errors is a org.apache.spark.rdd.FilteredRDD

errors.cache() // persist 到内存中

errors.count() // 触发action，计算errors有多少个，即ERROR的多少行

// Count errors mentioning MySQL:

errors.filter(_.contains("MySQL")).count() 

// Return the time fields of errors mentioning

// HDFS as an array (assuming time is field

// number 3 in a tab-separated format):

errors.filter(_.contains("HDFS"))

        .map(_.split('t')(3))

        .collect()

spark是一个org.apache.spark.SparkContext的实例，spark的应用从定义一个SparkContext开始：
textFile的定义如下： 

/**

* Read a text file from HDFS, a local file system (available on all nodes), or any

* Hadoop-supported file system URI, and return it as an RDD of Strings.

*/

def textFile(path: String, minPartitions: Int = defaultMinPartitions): RDD[String] = {

hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],

minPartitions).map(pair => pair._2.toString).setName(path)

}

hadoopFile创建了一个org.apache.spark.rdd.HadoopRDD，
而在HadoopRDD上调用map则生成了一个MappedRDD： 
 
/**

* Return a new RDD by applying a function to all elements of this RDD.

*/

def map[U: ClassTag](f: T => U): RDD[U] = 
new
 MappedRDD(
this
, sc.clean(f))

 errors.cache()并不会立即执行，它的作用是在RDD的计算完成后，将结果cache起来，以供以后的计算使用，
 这样的话可以加快以后运算的速度。
 errors.count() 就触发了一个action，这个时候就需要向集群提交job了：
 
/**

  * Return the number of elements in the RDD.

  */

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

 提交后，SparkContext会将runJob提交到DAGScheduler，DAGScheduler会将当前的DAG划分成Stage，
 然后生成TaskSet后通过TaskScheduler的submitTasks提交tasks，而这又会调用SchedulerBackend，
 SchedulerBackend会将这些任务发送到Executor去执行。