开发者学堂课程【大数据实时计算框架 Spark 快速入门：SparkPi 代码剖析2】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/100/detail/1677
SparkPi 代码剖析2
内容介绍：
一、SparkPi 代码剖析2
二、反射的方式来创建一个 TextInputFormat
一、SparkPi 代码剖析2
HadoopRDD 举例子
RDD.scala
236
final def partitions: Array[Partition]
239
partitions- getPartitions
HadoopRDD
149
override def getPartitions: Array[Partition]
185
val newInputFormat = ReflectionUtils.newInstance (inputFormatclass.asInstanceof[classl_]],co.asInstanceof[InputFormat[K, V]]
二、反射的方式来创建一个 TextInputFormat
inputFormatclass 哪里来的？??
这个地方可以查看sc.textFile()
SparkContext,scala
def textFile(
832
hadoopFile(path,classof[TextInputFormat],classof[Longwritable],classof[Text],minPartitions). map (pair = > pair._ 2. tostring). setName (path)
1006
def hadoopFile[K, V](
1016
new HadoopRDD(
this,
confBroadcast,
Some (setInputPathsFunc),
inputFormatclass,
keyclass,
valueclass,
minPartitions). setName (path)
inputFormatclass这个参数传个HadoopRDD之后，我们是不是就可以回到刚刚HadoopRDD里面去了
HadoopRDD.scala
199
val inputsplits- inputFormat. getsplits (jobConf, minpartitions)
通过getsplits()这个MapReduce框架里面的方法来获取到path路径所对应的split的Hosts getsplits()方法，我们可以通过TextInputFormat找到它继承的FileInputFormat来找到方法hadoop代码里面的
FileInputFormat.class
210
public Inputsplit[] getsplits (JobConf job, int numSplits) throws IOException (DagScheduler.scala
1553
rdd.dependencies. foreach {
casen: NarrowDependency [_] =>
for (inPart <- n.getparents (partition))
val locs- getPreferredLocsInternal(n.rdd,inPart,
visited)
if (locs!- Nil) {
returnlocs
Case
如果一直找到这个 stage 里面的最开始 RDD ,还是没有最近计算位置
那就返回 Nil
1565
Nil
1021
val tasks: seq[Task[_]] = try {
这里面就是会封装 Tasks
1502
这里面最终会把 Tasks 封装成 Taskset 传递给下游的 TaskScheduler
taskscheduler.submitTasks(newTaskset(
tasks. toArray, stage. id, stage. latest Info. attemptid, jobId, properties))submitTasks
如果要跟进去就要找到具体的 taskschedulerImpl
SparkContext.scala
2592
private def createTaskscheduler(
具体得看你用什么模式来运行的
2602
case "local"-
val scheduler = new TaskschedulerImpl (sc, MAX_ LOCAL_ TASK_ FAILURES, islocal- true)
TaskschedulerImpl.scala