sc.textFile("hdfs://....").flatMap(line =>line.split(" ")).map(w =>(w,1)).reduceByKey(_+_).foreach(println)
不使用reduceByKey
sc.textFile("hdfs://....").flatMap(l=>l.split(" ")).map(w=>(w,1)).groupByKey().map((p:(String,Iterable[Int]))=>(p._1，p._2.sum)).collect
步骤1：textFile先生成HadoopRDD,然后再通过map操作生成MappedRDD.
结果:res0:org.apache.spark.rdd.RDD[String] = MappedRDD[1] at textFile at :13
步骤2：val split = line =>line.split(” “)).flatMap(line => line.split(” “)) flatMap将原来的MappedRDD转换为FlatMappedRDD
步骤3：val wordCount = split.map(w =>(w,1)) 利用w生成相应的键值对，上一步的FlatMappedRDD被转换为MappedRDD
步骤4：val reduce = wordCount.reduceByKey(_+_)
步骤5：reduce.foreach(println) 触发执行
在执行foreach时，调用了runJob函数，实现了重载。 Final RDD和作用于RDD上的Function。 然后读取Finall RDD的分区数，通过allowLocal来表示是否在Standalone模式下执行。
从spark-shell到sparkContext的创建的调用路径:
spark-shell -> spark-submit ->spark-class->sparkSubmit.main ->SparkILoop -> createSparkContext
SpackContext初始化过程中 传入的入参是SparkConf
一、根据初始化生成SparkConf,再根据SparkConf来创建SparkEnv.
二、创建TaskScheduler,根据Spark的运行模式选择相应的SchedulerBackend，同时启动TaskScheduler
private[spark] var taskScheduler = SparkContext.createTaskScheduler(this,master,appName)
taskScheduler.start()
createTaskScheduler最为关键，根据master环境变量来判断Spark当前的部署方式，从而生成相应的SchedulerBackend的不同子类。taskScheduler.start的目的是启动相应的SchedulerBackend.
三、从上一步创建的taskScheduler实例为入参创建DAGScheduler并启动运行。
private[spark] var dagScheduler = new DAGScheduler(taskScheduler)
dagScheduler.start()
四、启动WebUI.
ui.start()
