关于spark运行原理的请教

示例代码如下：
[mw_shl_code=scala,true]
    val conf = new SparkConf().setAppName("CollectFemaleInfo")

    // Initializing Spark

    val sc = new SparkContext(conf)

    // Read data. This code indicates the data path that the input parameter args(0) specifies.

    val text = sc.textFile(args(0))

    // Filter the data information about the time that female netizens spend online.

    val data = text.filter(_.contains("female"))

    // Aggregate the time that each female netizen spends online

    val femaleData:RDD[(String,Int)] = data.map{line =>

      val t= line.split(',')

      (t(0),t(2).toInt)

    }.reduceByKey(_ + _)

    // Filter the information about female netizens who spend more than 2 hours online, and export the results

    val result = femaleData.filter(line => line._2 > 120)

    result.collect().map(x => x._1 + ',' + x._2).foreach(println)

    sc.stop()[/mw_shl_code]

问题：
1、个人理解    val sc = new SparkContext(conf) 会创建一个driver，后面的全是executor,Driver代码只是在一台机器跑，剩下的代码在多台机器跑？

2、从val text = sc.textFile(args(0))后，我理解自此之后的每一个算子都是在各个节点分布式执行的。且算出的结果保存在各个节点的RDD中。此处有一个疑问，针对每一个节点，代码是不是顺序执行的呢？ 如果有节点A和B，针对A和B的某一个partition，A执行到reduceByKey，B执行到femaleData.filter，请问有这种可能性吗？  不太理解各个算子在各个节点的执行顺序？

谢谢啊

1、我理解的driver也是这个意思，不过不是所有代码都需要“跑”！多数是生成了DAG图
2、代码跑与不跑还要看代码是那种类型的，分Transformations和Actions，另外分布式必然存在执行顺序问题，但是在RDD真正执行前会产生DAG描述图，个别RDD丢失也会根据关系重新产生，所以不用担心丢失与错乱问题
下图是我本地测试的截图，DAG已经描述的很清楚流程与关系

1、个人理解    val sc = new SparkContext(conf) 会创建一个driver，后面的全是executor,Driver代码只是在一台机器跑，剩下的代码在多台机器跑？

第一也是赞同，如上图所示

2、从val text = sc.textFile(args(0))后，我理解自此之后的每一个算子都是在各个节点分布式执行的。且算出的结果保存在各个节点的RDD中。此处有一个疑问，针对每一个节点，代码是不是顺序执行的呢？ 如果有节点A和B，针对A和B的某一个partition，A执行到reduceByKey，B执行到femaleData.filter，请问有这种可能性吗？  不太理解各个算子在各个节点的执行顺序？
算子分两类，一类Transformation，一类Action，只有Action算子会触发SparkContext提交作业。楼主说的不够清晰，这里给出下面图示

如下图A中有三个分区，到B有三个分区，A到B执行map算子这是stage1
stage2中有CDEF，C到D执行map算子，D到F执行reduce算子。这里面正是楼主的问题，如果C的分区1partition1 在执行map，分区2partition2是否会执行到reduce.这个是有可能的。但是reduce执行完毕，这时候他就需要等待了，甚至还需要等待E到F，同时还需要等待stage1中的算子全部执行完毕，执行join操作

arsenduan 发表于 2017-3-22 17:25
1、个人理解    val sc = new SparkContext(conf) 会创建一个driver，后面的全是executor,Driver代码只是在 ...

谢谢。 我理解partition是各个算子执行的基本单位，各个partition数据处理时候 互不干扰 不知对不对

evababy 发表于 2017-3-22 14:10
1、我理解的driver也是这个意思，不过不是所有代码都需要“跑”！多数是生成了DAG图
2、代码跑与不跑还要 ...

您好，谢谢您的回复

其实我对一个概念一直很含糊，
1、什么是DAG描述图？  有没有具体的表现形式呢  好抽象  个人理解DAGScheduler模块介入运算，计算RDD之间的依赖关系。RDD之间的依赖关系就形成了DAG 。

2、我理解一般一个job 有一个或者多个stage，一个stage有一个或多个算子。 查找资料如下解释：
每一个JOB被分为多个Stage，划分Stage的一个主要依据是当前计算因子的输入是否是确定的，如果是则将其分在同一个Stage，避免多个Stage之间的消息传递开销


请问什么是输入是确定的呢？


谢谢啊




参考的是下面的文章：
http://blog.csdn.net/shadow_mi/article/details/51821613

Wyy_Ck 发表于 2017-3-23 11:14
您好，谢谢您的回复

其实我对一个概念一直很含糊，

客气了，我研究spark时间也不久1、DAG本身就是一个抽象的关系描述，标明数据的来源
2、个人理解，思路不是很清晰，大概说明的层级的关系。
个人对RDD\stages的理解：
1、存在多级RDD，从上层开始顺序执行，每个RDD都产生了一个stages
2、而RDD可以由一个action或一个、多个Transformation组成
3、不会推导出“不存在直接关系的同一个RDD”是否重复执行

[mw_shl_code=scala,true]val l1=l.map(a => (a,1))
val l2=l.map(a => (a,2))
val l3=l1.join(l2)[/mw_shl_code]
三个语句都不是action类型的，所以生成了三个rdd
l1: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[31] at map at <console>:26
l2: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[32] at map at <console>:26
l3: org.apache.spark.rdd.RDD[(String, (Int, Int))] = MapPartitionsRDD[35] at join at <console>:30

测试:1：执行l1.collect
会发现DAG只有文件到MAP的操作

测试2：执行l3.collec会发现出现2个MAP一个collect，但是DAG中并没有MAP相关的描述



测试3：重复l3.collect会发现只有一个collect任务，不过发现ID直接从20跳到23（由于是在同一个shell下，RDD临时缓存吧）


在复杂一些，为了避免内存溢出吧map改简单点
[mw_shl_code=scala,true]val l1=l.map(a => (a,1))
val l2=l.map(a => (a,1))
val l3=l1.join(l2)
val l4=l1.join(l2)
val l5=l3.join(l4)[/mw_shl_code]

测试4：l5.collect

存在5个任务，collect更为复杂，依然没有MAP，且执行了2次（每次是2个MAP）