2019独角兽企业重金招聘Python工程师标准>>> 
本文以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执行过程如下:

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