本文以WordCount为例, 画图说明spark程序的执行过程

WordCount就是统计一段数据中每个单词出现的次数, 例如hello spark hello you 这段文本中hello出现2次, spark出现1次, you出现1次. 先上完整代码:

object WordCount {  def main(args: Array[String]) {    val conf = new SparkConf().setAppName("WordCount");    val sc = new SparkContext(conf)      val lines = sc.textFile("hdfs://xxx:9000/spark.txt", 3);     val words = lines.flatMap { line => line.split("\s+") }       val pairs = words.map { word => (word, 1) }       val wordCounts = pairs.reduceByKey { _ + _ }    wordCounts.foreach(wordCount => println(wordCount._1 + " appeared " + wordCount._2 + " times."))    }}

上面几行代码就把hdfs上的spark.txt中每个单词出现的个数计算完成.

Spark集群的执行单位是Application，任何提交的任务都会产生一个Application。一个Application只会关联上一个Spark上下文，也就是SparkContext。构建SparkContext时可以传入Spark相关配置，也就是SparkConf，它可以用来指定Application的名称，任务需要的CPU核数/内存大小，调优需要的配置等等. 以下两行创建了SparkContext:

val conf = new SparkConf().setAppName("WordCount");val sc = new SparkContext(conf)

创建完SparkContext之后, spark.txt的文件数如何被spark处理的呢,让我们一起看一下:

首先我们假设spark.txt在hdfs上对应着3个文件,文件内容都一样,sc.textFile("hdfs://xxx:9000/spark.txt", 3)也执行了最小分区数为3. 然后wordcount执行过程如下:



说明:

1. 绿,红,黄色箭头的地方发生了`Shuffer,把整个任务分成了2个Stage(2个蓝色虚线框)
2. 红色虚线框代表一个Partition窄依赖(每个分区只被子RDD的一个分区所使用)的运行过程, 多个partition是并行执行的
3. reduceByKey会先把每个Partition中的数据预聚合(groupByKey不会)
4. Stage中的数据都是在内存中,不像MapReduce会频繁写磁盘,速度很快.
5. 补充:其实textFile,flatMap,map,reduceByKey等transformation操作都是lazy的,程序执行到这里不会立即执行,只有再触发action操作的时候才会执行,此例中为wordCounts.foreach这个action操作.

--Posted from Rpc