Spark中的Scheduler
scheduler分成两个类型,一个是TaskScheduler与其实现,一个是DAGScheduler。
TaskScheduler:主要负责各stage中传入的task的执行与调度。
DAGScheduler:主要负责对JOB中的各种依赖进行解析,根据RDD的依赖生成stage并通知TaskScheduler执行。
实例生成
TaskScheduler实例生成:
scheduler实例生成,我目前主要是针对on yarn的spark进行的相关分析,
在appmaster启动后,通过调用startUserClass()启动线程来调用用户定义的spark分析程序。
传入的第一个参数为appmastername(master),可传入的如:yarn-cluster等。
在用户定义的spark分析程序中,生成SparkContext实例。
通过SparkContext.createTaskScheduler函数。如果是yarn-cluster,生成YarnClusterScheduler实例。
此部分生成的scheduler为TaskScheduler实例。
defthis(sc: SparkContext) = this(sc, new Configuration())
同时YarnClusterSchduler实现TaskSchedulerImpl。
defthis(sc: SparkContext) = this(sc, sc.conf.getInt("spark.task.maxFailures", 4))
生成TaskScheduler中的SchedulerBackend属性引用,yarn-cluster为CoarseGrainedSchedulerBackend
valbackend = new CoarseGrainedSchedulerBackend(scheduler, sc.env.actorSystem)
scheduler.initialize(backend)
DAGScheduler实例生成:
class DAGScheduler(
taskSched: TaskScheduler,
mapOutputTracker: MapOutputTrackerMaster,
blockManagerMaster: BlockManagerMaster,
env: SparkEnv)
extends Logging {
defthis(taskSched: TaskScheduler) {
this(taskSched, SparkEnv.get.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster],
SparkEnv.get.blockManager.master, SparkEnv.get)
}
taskSched.setDAGScheduler(this)
scheduler调度过程分析
1.rdd执行action操作,如saveAsHadoopFile
2.调用SparkContext.runJob
3.调用DAGScheduler.runJob-->此函数调用submitJob,并等job执行完成。
Waiter.awaitResult()中通过_jobFinished检查job运行是否完成,如果完成,此传为true,否则为false.
_jobFinished的值通过resultHandler函数,每调用一次finishedTasks的值加一,
如果finishedTasks的个数等于totalTasks的个数时,表示完成。或者出现exception.
def runJob[T, U: ClassTag](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: String,
allowLocal: Boolean,
resultHandler: (Int, U) => Unit,
properties: Properties = null)
{
valwaiter = submitJob(rdd, func, partitions, callSite, allowLocal, resultHandler, properties)
waiter.awaitResult() match {
case JobSucceeded => {}
case JobFailed(exception: Exception, _) =>
logInfo("Failed to run " + callSite)
throwexception
}
}
4.调用DAGScheduler.submitJob函数,
部分代码:生成JobWaiter实例,并传入此实例,发送消息,调用JobSubmitted事件。并返回waiter实例。
JobWaiter是JobListener的实现。
valwaiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
eventProcessActor ! JobSubmitted(
jobId, rdd, func2, partitions.toArray, allowLocal, callSite, waiter, properties)
waiter
5.处理DAGScheduler的JobSubmitted事件消息,通过processEvent处理消息接收的事件。
def receive = {
caseevent: DAGSchedulerEvent =>
logTrace("Got event of type " + event.getClass.getName)
if (!processEvent(event)) {
submitWaitingStages()
} else {
resubmissionTask.cancel()
context.stop(self)
}
}
}))
6. processEvent函数中处理JobSubmitted事件部分代码:
case JobSubmitted(jobId, rdd, func, partitions, allowLocal, callSite, listener, properties) =>
varfinalStage: Stage = null
try {
生成stage实例,stage的id通过nextStageId的值加一得到,task的个数就是partitions的分区个数,
根据job对应的rdd,得到如果parent rdd是shuffle的rdd时生成ShuffleMapStage,通过getParentStages函数,
此处去拿到parent rdd时,如果current rdd的parent rdd不是shuffle,递归调用parent rdd,
如果parend rdd中没有shuffle的rdd,不生成新的stage,否则有多少个,生成多少个。此处是处理DAG类的依赖
finalStage = newStage(rdd, partitions.size, None, jobId, Some(callSite))
} catch {
casee: Exception =>
logWarning("Creating new stage failed due to exception - job: " + jobId, e)
listener.jobFailed(e)
returnfalse
}
生成ActiveJob实例。设置numFinished的值为0,表示job中有0个完成的task.
设置所有task个数的array finished.并把所有元素的值设置为false.把JobWaiter当listener传入ActiveJob.
valjob = new ActiveJob(jobId, finalStage, func, partitions, callSite, listener, properties)
对已经cache过的TaskLocation进行清理。
clearCacheLocs()
logInfo("Got job " + job.jobId + " (" + callSite + ") with " + partitions.length +
" output partitions (allowLocal=" + allowLocal + ")")
logInfo("Final stage: " + finalStage + " (" + finalStage.name + ")")
logInfo("Parents of final stage: " + finalStage.parents)
logInfo("Missing parents: " + getMissingParentStages(finalStage))
如果runJob时传入的allowLocal的值为true,同时没有需要shuffle的rdd,同时partitions的长度为1,
也就是task只有一个,直接在local运行此job..通过runLocallyWithinThread生成一个线程来执行。
if (allowLocal && finalStage.parents.size == 0 && partitions.length == 1) {
// Compute very short actions like first() or take() with no parent stages locally.
listenerBus.post(SparkListenerJobStart(job, Array(), properties))
通过ActiveJob中的func函数来执行job的运行,此函数在rdd的action调用时生成定义,
如saveAsHadoopFile(saveAsHadoopDataset)中的定义的内部func,writeToFile函数。
完成函数执行后,调用上面提到的生成的JobWaiter.taskSucceeded函数。
runLocally(job)
} else {
否则有多个partition也就是有多个task,或者有shuffle的情况,
idToActiveJob(jobId) = job
activeJobs += job
resultStageToJob(finalStage) = job
listenerBus.post(SparkListenerJobStart(job, jobIdToStageIds(jobId).toArray, properties))
调用DAGScheduler.submitStage函数。
submitStage(finalStage)
}
7.DAGScheduler.submitStage函数:递归函数调用,
如果stage包含parent stage(shuffle的情况)把stage设置为waiting状态,等待parent stage执行完成才进行执行。
privatedef submitStage(stage: Stage) {
valjobId = activeJobForStage(stage)
if (jobId.isDefined) {
logDebug("submitStage(" + stage + ")")
如果RDD的Dependency的RDD还没有执行完成,等待Dependency执行完成后当前的RDD才能进行执行操作。
if (!waiting(stage) && !running(stage) && !failed(stage)) {
根据stage中rdd的Dependency,检查是否需要生成新的stage,如果是ShuffleDependency,会生成新的ShuffleMapStage
此处去拿到parent rdd时,如果current rdd的parent rdd不是shuffle,递归调用parent rdd,
如果parend rdd中没有shuffle的rdd,不生成新的stage,否则有多少个,生成多少个。此处是处理DAG类的依赖
valmissing = getMissingParentStages(stage).sortBy(_.id)
logDebug("missing: " + missing)
如果没有RDD中的shuffle的Dependency,也就是RDD之间都是NarrowDependency的Dependency
表示所有的Dependency都在map端本地执行。
if (missing == Nil) {
logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
submitMissingTasks(stage, jobId.get)
running += stage
} else {
如果RDD有Dependency,先执行parent rdd的stage操作。此处是递归函数调用
for (parent <- missing) {
submitStage(parent)
}
waiting += stage
}
}
} else {
abortStage(stage, "No active job for stage " + stage.id)
}
}
8. DAGScheduler.submitMissingTask的执行流程:
privatedef submitMissingTasks(stage: Stage, jobId: Int) {
logDebug("submitMissingTasks(" + stage + ")")
// Get our pending tasks and remember them in our pendingTasks entry
valmyPending = pendingTasks.getOrElseUpdate(stage, new HashSet)
myPending.clear()
vartasks = ArrayBuffer[Task[_]]()
如果stage是shuffle的rdd,迭代stage下的的所有partition,根据partition与对应的TaskLocation
生成ShuffleMapTask.添加到task列表中。
if (stage.isShuffleMap) {
for (p <- 0 until stage.numPartitionsif stage.outputLocs(p) == Nil) {
vallocs = getPreferredLocs(stage.rdd, p)
tasks += new ShuffleMapTask(stage.id, stage.rdd, stage.shuffleDep.get, p, locs)
}
} else {
否则表示stage是非shuffle的rdd,此是是执行完成后直接返回结果的stage,生成ResultTask实例。
由于是ResultTask,因此需要传入定义的func,也就是如何处理结果返回
// This is a final stage; figure out its job's missing partitions
valjob = resultStageToJob(stage)
for (id <- 0 until job.numPartitionsif !job.finished(id)) {
valpartition = job.partitions(id)
vallocs = getPreferredLocs(stage.rdd, partition)
tasks += new ResultTask(stage.id, stage.rdd, job.func, partition, locs, id)
}
}
valproperties = if (idToActiveJob.contains(jobId)) {
idToActiveJob(stage.jobId).properties
} else {
//this stage will be assigned to "default" pool
null
}
// must be run listener before possible NotSerializableException
// should be "StageSubmitted" first and then "JobEnded"
listenerBus.post(SparkListenerStageSubmitted(stageToInfos(stage), properties))
if (tasks.size > 0) {
// Preemptively serialize a task to make sure it can be serialized. We are catching this
// exception here because it would be fairly hard to catch the non-serializable exception
// down the road, where we have several different implementations for local scheduler and
// cluster schedulers.
try {
SparkEnv.get.closureSerializer.newInstance().serialize(tasks.head)
} catch {
casee: NotSerializableException =>
abortStage(stage, "Task not serializable: " + e.toString)
running -= stage
return
}
logInfo("Submitting " + tasks.size + " missing tasks from " + stage + " (" + stage.rdd + ")")
myPending ++= tasks
logDebug("New pending tasks: " + myPending)
生成TaskSet实例,把stage中要执行的Task列表传入,同时把stage对应的ActiveJob也传入。
通过TaskScheduler的实现,调用submitTasks函数,YarnClusterScheduler(TaskSchedulerImpl)
taskSched.submitTasks(
new TaskSet(tasks.toArray, stage.id, stage.newAttemptId(), stage.jobId, properties))
stageToInfos(stage).submissionTime = Some(System.currentTimeMillis())
} else {
logDebug("Stage " + stage + " is actually done; %b %d %d".format(
stage.isAvailable, stage.numAvailableOutputs, stage.numPartitions))
running -= stage
}
}
9.TaskSchedulerImpl.submitTasks函数流程分析:
通过传入的TaskSet,得到要执行的tasks列表,并生成TaskSetmanager实例,
同时把实例添加到的schedulableBuilder(FIFOSchedulableBuilder/FairSchedulableBuilder)队列中。
关于TaskSetManager实例可参见后面的分析。
overridedef submitTasks(taskSet: TaskSet) {
valtasks = taskSet.tasks
logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
this.synchronized {
valmanager = new TaskSetManager(this, taskSet, maxTaskFailures)
activeTaskSets(taskSet.id) = manager
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
taskSetTaskIds(taskSet.id) = new HashSet[Long]()
定期检查task的执行消息是否被生成执行。如果task被分配执行,关闭此线程。否则一直给出提示.
if (!isLocal && !hasReceivedTask) {
starvationTimer.scheduleAtFixedRate(new TimerTask() {
overridedef run() {
if (!hasLaunchedTask) {
logWarning("Initial job has not accepted any resources; " +
"check your cluster UI to ensure that workers are registered " +
"and have sufficient memory")
} else {
this.cancel()
}
}
}, STARVATION_TIMEOUT, STARVATION_TIMEOUT)
}
hasReceivedTask = true
}
通过SchedulerBackend的实现CoarseGrainedSchedulerBackend.reviceOffers发起执行处理操作。
backend.reviveOffers()
}
9.1TaskSetManager的实例生成:
private[spark] class TaskSetManager(
sched: TaskSchedulerImpl,
valtaskSet: TaskSet,
valmaxTaskFailures: Int,
clock: Clock = SystemClock)
extendsSchedulablewith Logging
...........................
for (i <- (0 until numTasks).reverse) {
addPendingTask(i)
}
关于addPendingTask的定义:此睦传入的readding的值为false.
privatedef addPendingTask(index: Int, readding: Boolean = false) {
// Utility method that adds `index` to a list only if readding=false or it's not already there
内部定义的addTo方法。
defaddTo(list: ArrayBuffer[Int]) {
if (!readding || !list.contains(index)) {
list += index
}
}
varhadAliveLocations = false
迭代所有的要执行的task,并通过task的TaskLocation检查执行的节点级别。添加到相应的pendingTask容器中
for (loc <- tasks(index).preferredLocations) {
for (execId <- loc.executorId) {
检查TaskSchedulerImpl.activeExecutorIds的活动的worker的executor是否存在,
如果是第一个执行的RDD时,此时activeExecutorIds容器的的值为空,当第一个RDD中有TASK在此executor中执行过后,
会把executor的id添加到activeExecutorIds容器中。
第一个RDD的stage执行时,此部分不执行,但第二个stage执行时,可最大可能的保证task在PROCESS_LOCAL的执行。
if (sched.isExecutorAlive(execId)) {
addTo(pendingTasksForExecutor.getOrElseUpdate(execId, new ArrayBuffer))
hadAliveLocations = true
}
}
if (sched.hasExecutorsAliveOnHost(loc.host)) {
如果在TaskSchedulerImpl的executorsByHost容器中包含此host,在pendingTasksForHost 中添加对应的task.
TaskSchedulerImpl.executorsByHost容器的值在每一个worker注册时
通过向CoarseGrainedSchedulerBackend.DriverActor发送RegisterExecutor事件消息。
通过makeOffers()-->TaskSchedulerImpl.resourceOffers把host添加到executorsByHost容器中。
addTo(pendingTasksForHost.getOrElseUpdate(loc.host, new ArrayBuffer))
通过调用YarnClusterScheduler.getRackForHost得到host对应的rack,
并在rack的pending容器中添加对应的task个数和。
for (rack <- sched.getRackForHost(loc.host)) {
addTo(pendingTasksForRack.getOrElseUpdate(rack, new ArrayBuffer))
}
hadAliveLocations = true
}
}
如果上面两种情况都没有添加到容器中pendingTasksWithNoPrefs。
if (!hadAliveLocations) {
// Even though the task might've had preferred locations, all of those hosts or executors
// are dead; put it in the no-prefs list so we can schedule it elsewhere right away.
addTo(pendingTasksWithNoPrefs)
}
在TaskSetManager实例生成是,把所有task的个数都添加到allPendingTasks 容器中
if (!readding) {
allPendingTasks += index // No point scanning this whole list to find the old task there
}
}
.............................
得到可选择的LocalityLevel级别。
valmyLocalityLevels = computeValidLocalityLevels()
vallocalityWaits = myLocalityLevels.map(getLocalityWait) // Time to wait at each level
以下代码是 computeValidLocalityLevels的定义,主要根据各种locality中pending的容器中是否有值。
生成当前stage中的task执行可选择的Locality级别。
privatedef computeValidLocalityLevels(): Array[TaskLocality.TaskLocality] = {
import TaskLocality.{PROCESS_LOCAL, NODE_LOCAL, RACK_LOCAL, ANY}
vallevels = new ArrayBuffer[TaskLocality.TaskLocality]
if (!pendingTasksForExecutor.isEmpty && getLocalityWait(PROCESS_LOCAL) != 0) {
levels += PROCESS_LOCAL
}
if (!pendingTasksForHost.isEmpty && getLocalityWait(NODE_LOCAL) != 0) {
levels += NODE_LOCAL
}
if (!pendingTasksForRack.isEmpty && getLocalityWait(RACK_LOCAL) != 0) {
levels += RACK_LOCAL
}
levels += ANY
logDebug("Valid locality levels for " + taskSet + ": " + levels.mkString(", "))
levels.toArray
}
}
以下代码是getLocalityWait的定义代码:此函数主要是定义每一个Task在此Locality级别中执行的等待时间。
也就是scheduler调度在传入的Locality级别时所花的时间是否超过指定的等待时间,
如果超过表示需要放大Locality的查找级别。
privatedef getLocalityWait(level: TaskLocality.TaskLocality): Long = {
valdefaultWait = conf.get("spark.locality.wait", "3000")
level match {
case TaskLocality.PROCESS_LOCAL =>
conf.get("spark.locality.wait.process", defaultWait).toLong
case TaskLocality.NODE_LOCAL =>
conf.get("spark.locality.wait.node", defaultWait).toLong
case TaskLocality.RACK_LOCAL =>
conf.get("spark.locality.wait.rack", defaultWait).toLong
case TaskLocality.ANY =>
0L
}
}
10.SchedulerBackend.reviveOffers()的调度处理流程:
SchedulerBackend的实现为CoarseGrainedSchedulerBackend。
overridedef reviveOffers() {
driverActor ! ReviveOffers
}
以上代码发CoarseGrainedSchedulerBackend内部的DriverActor发送消息,处理ReviveOffers事件。
case ReviveOffers =>
makeOffers()
................
def makeOffers() {
见下面的launchTasks与resourceOffers函数
launchTasks(scheduler.resourceOffers(
executorHost.toArray.map {case (id, host) => new WorkerOffer(id, host, freeCores(id))}))
}
调用TaskSchedulerImpl.resourceOffers并传入注册的worker中executorid与host的kv array.
def resourceOffers(offers: Seq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {
SparkEnv.set(sc.env)
// Mark each slave as alive and remember its hostname
for (o <- offers) {
executorIdToHost(o.executorId) = o.host
此部分主要是在worker注册时executorsByHost中还不存在时会执行,
if (!executorsByHost.contains(o.host)) {
executorsByHost(o.host) = new HashSet[String]()
executorGained(o.executorId, o.host)
}
}
offers表示有多少个注册的worker的executor,根据每一个worker中可能的cpu core个数生成可执行的task个数。
// Build a list of tasks to assign to each worker
valtasks = offers.map(o => new ArrayBuffer[TaskDescription](o.cores))
可分配的cpu个数,由此处可以看出每一个任务分配时最好按每个worker能分配的最大cpu core个数来分配。
valavailableCpus = offers.map(o => o.cores).toArray
得到队列中的所有的TaskSetManager列表。
valsortedTaskSets = rootPool.getSortedTaskSetQueue()
for (taskSet <- sortedTaskSets) {
logDebug("parentName: %s, name: %s, runningTasks: %s".format(
taskSet.parent.name, taskSet.name, taskSet.runningTasks))
}
计算task的Locality级别,launchedTask=false表示需要放大Locality的级别。
// Take each TaskSet in our scheduling order, and then offer it each node in increasing order
// of locality levels so that it gets a chance to launch local tasks on all of them.
varlaunchedTask = false
计算task的Locality,此处是一个for的迭代调用,先从taskset列表中拿出一个tasetset,
子迭代是从PROCESS_LOCAL开始迭代locality的级别。
for (taskSet <- sortedTaskSets; maxLocality <- TaskLocality.values) {
do {
launchedTask = false
迭代调用每一个worker的值,从每一个worker中在taskset中选择task的执行级别,生成TaskDescription
for (i <- 0 until offers.size) {
得到迭代出的worker的executorid与host
valexecId = offers(i).executorId
valhost = offers(i).host
通过TaskSetManager.resourceOffer选择一个执行级别,通过此函数选择Locality级别时,
不能超过传入的maxLocality,每次生成一个task,
for (task <- taskSet.resourceOffer(execId, host, availableCpus(i), maxLocality)) {
每次生成一个task,把生成的task添加到上面的tasks列表中。
tasks(i) += task
valtid = task.taskId
taskIdToTaskSetId(tid) = taskSet.taskSet.id
taskSetTaskIds(taskSet.taskSet.id) += tid
taskIdToExecutorId(tid) = execId
设置当前executorid设置到activeExecutorIds列表中,当有多个依赖的stage执行时,
第二个stage在submitTasks时,生成TaskSetManager时,会根据的activeExecutorIds值,
在pendingTasksForExecutor中生成等执行的PROCESS_LOCAL的pending tasks.
activeExecutorIds += execId
把executor对应的host记录到executorsByHost容器中。
executorsByHost(host) += execId
当前worker中可用的cpu core的值需要减去一,这样能充分保证一个cpu core执行一个task
availableCpus(i) -= 1
这个值用来检查是否在当前的Locality级别中接着执行其它的task的分配,
如果这个值为true,不放大maxLocality的级别,从下一个worker中接着分配剩余的task
launchedTask = true
}
}
} while (launchedTask)
}
if (tasks.size > 0) {
设置hasLaunchedTask的值为true,表示task的执行分配完成,在上面提到过的检查线程中对线程执行停止操作。
hasLaunchedTask = true
}
returntasks
}
10.1 TaskSetManager.resourceOffer流程分析
def resourceOffer(
execId: String,
host: String,
availableCpus: Int,
maxLocality: TaskLocality.TaskLocality)
: Option[TaskDescription] =
{
如果完成的task个数小于要生成的总task个数,同时当前cpu可用的core个数和大于或等于一个配置的,默认1
if (tasksSuccessful < numTasks && availableCpus >= CPUS_PER_TASK) {
valcurTime = clock.getTime()
通过现在执行task分配的时间减去上一次并从currentLocalityIndex的下标开始,
取出locality对应的task分配等待时间,如果时间超过了此配置,把下标值加一,
找到下一个locality的配置时间,按这方式找,直到找到ANY的值,具体可见下面的此方法说明
varallowedLocality = getAllowedLocalityLevel(curTime)
如果通过的locality的级别超过了传入的最大locality级别,把级别设置为传入的最大级别
if (allowedLocality > maxLocality) {
allowedLocality = maxLocality // We're not allowed to search for farther-away tasks
}
findTask主要是从对应的pending的列表中根据对应的Locality拿到对应的task的下标,在TaskSet.tasks中的下标。
findTask(execId, host, allowedLocality) match {
case Some((index, taskLocality)) => {
// Found a task; do some bookkeeping and return a task description
valtask = tasks(index)
valtaskId = sched.newTaskId()
// Figure out whether this should count as a preferred launch
logInfo("Starting task %s:%d as TID %s on executor %s: %s (%s)".format(
taskSet.id, index, taskId, execId, host, taskLocality))
// Do various bookkeeping
copiesRunning(index) += 1
valinfo = new TaskInfo(taskId, index, curTime, execId, host, taskLocality)
taskInfos(taskId) = info
taskAttempts(index) = info :: taskAttempts(index)
把分配此task的locality级别拿到对应的下标,并重新设置下标的值。
// Update our locality level for delay scheduling
currentLocalityIndex = getLocalityIndex(taskLocality)
把这次的task的分配时间设置成最后一次分配时间。
lastLaunchTime = curTime
// Serialize and return the task
valstartTime = clock.getTime()
// We rely on the DAGScheduler to catch non-serializable closures and RDDs, so in here
// we assume the task can be serialized without exceptions.
valserializedTask = Task.serializeWithDependencies(
task, sched.sc.addedFiles, sched.sc.addedJars, ser)
valtimeTaken = clock.getTime() - startTime
addRunningTask(taskId)
logInfo("Serialized task %s:%d as %d bytes in %d ms".format(
taskSet.id, index, serializedTask.limit, timeTaken))
valtaskName = "task %s:%d".format(taskSet.id, index)
如果是第一次执行,通过DAGScheduler.taskStarted发送BeginEvent事件。
if (taskAttempts(index).size == 1)
taskStarted(task,info)
return Some(new TaskDescription(taskId, execId, taskName, index, serializedTask))
}
case _ =>
}
}
None
}
根据超时时间配置,如果这次分配task的时间减去上次task分配的时间超过了locality分配等待的配置时间,
把locality的级别向上移动一级,并重新比对时间,拿到不超时的locality级别或ANY的级别。
privatedef getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
currentLocalityIndex < myLocalityLevels.length - 1)
{
下标值加一,也就是把当前的Locality的级别向上放大一级。
// Jump to the next locality level, and remove our waiting time for the current one since
// we don't want to count it again on the next one
lastLaunchTime += localityWaits(currentLocalityIndex)
currentLocalityIndex += 1
}
myLocalityLevels(currentLocalityIndex)
}
DAGScheduler中处理BeginEvent事件:
case BeginEvent(task, taskInfo) =>
for (
job <- idToActiveJob.get(task.stageId);
stage <- stageIdToStage.get(task.stageId);
stageInfo <- stageToInfos.get(stage)
) {
if (taskInfo.serializedSize > TASK_SIZE_TO_WARN * 1024 &&
!stageInfo.emittedTaskSizeWarning) {
stageInfo.emittedTaskSizeWarning = true
logWarning(("Stage %d (%s) contains a task of very large " +
"size (%d KB). The maximum recommended task size is %d KB.").format(
task.stageId, stageInfo.name, taskInfo.serializedSize / 1024, TASK_SIZE_TO_WARN))
}
}
listenerBus.post(SparkListenerTaskStart(task, taskInfo))
11. CoarseGrainedSchedulerBackend.launchTasks流程
执行task的执行,发送LaunchTask事件处理消息
def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
for (task <- tasks.flatten) {
freeCores(task.executorId) -= 1
根据worker注册时的actor,向此actor发送LaunchTask事件。
executorActor(task.executorId) ! LaunchTask(task)
}
}
12.启动task,由于是on yarn的模式,worker的actor在CoarseGrainedExecutorBackend.
处理代码如下:
case LaunchTask(taskDesc) =>
logInfo("Got assigned task " + taskDesc.taskId)
if (executor == null) {
logError("Received LaunchTask command but executor was null")
System.exit(1)
} else {
executor.launchTask(this, taskDesc.taskId, taskDesc.serializedTask)
}
.............................
通过Executor启动task的执行。
其它actor的消息处理与task的具体执行与shuffle后面分析,这里先不做细的说明。
相关推荐
在Dolphinscheduler中,你需要为Spark添加相应的环境变量,包括`SPARK_HOME`、`HADOOP_CONF_DIR`等,以便Dolphinscheduler能够正确地执行Spark任务。 Hive的集成允许Dolphinscheduler执行SQL查询并处理Hive表。在...
1. **Spark已安装**:确保您的系统中已正确安装Spark。 2. **运行模式**:Spark可以在多种模式下运行,如local、local-cluster等。对于本次分析,我们关注的是local-cluster模式。可以通过`MASTER=local[1,2,1024] ...
`org.apache.spark.scheduler.TaskScheduler`接口及其实现类如`org.apache.spark.scheduler.cluster.YarnClusterScheduler`和`org.apache.spark.scheduler.local.LocalScheduler`展示了如何根据不同的部署模式进行...
在源码中,`org.apache.spark.scheduler.DAGScheduler`是负责这个过程的关键类。 3. **内存管理**:Spark的一大优势是其高效的内存管理。它将数据存储在内存中,减少了磁盘I/O,提高了计算速度。`org.apache.spark....
Task Scheduler是Spark中的关键模块,它负责将DAG分解为一系列可执行的任务。Task Scheduler分为两层:Stage划分和Task分配。Stage是Spark作业的最小调度单元,由一系列相互依赖的Task组成,同一Stage内的Task可以...
在本文中,我们将探讨如何利用Dolphinscheduler构建基于Hadoop、Spark、Hive的工作流模板,以及涉及到的数据同步与运算过程。 首先,我们关注的是“HIVE建表”这一环节。Hive是Hadoop生态系统中的一个数据仓库工具...
基于DolphinScheduler构建大数据调度平台,课程会带大家构建DolphinScheduler大数据调度平台,实战讲解多种任务调度配置,会基于案例讲解DolphinScheduler使用,让大家在实战中掌握DolphinScheduler。
Spark的核心在于其内存计算模型,它能够在内存中存储中间结果,避免频繁读写硬盘,极大地提高了数据处理速度。此外,Spark与Scala的紧密结合使得开发更为便捷,开发者可以像处理本地集合一样处理分布式数据集。Spark...
在术语定义中,Application指的是用户编写的Spark应用程序,它由运行main()函数的Driver和分布在集群中多个节点上运行的Executor组成。Driver创建SparkContext以准备运行环境,SparkContext负责与ClusterManager通信...
4. **网络通信**:Akka在Spark中的作用,以及如何实现高效的节点间通信。 5. **性能优化**:Tungsten项目如何提升DataFrame的性能,以及Code Generation技术。 6. **错误处理**:容错逻辑的实现,如何通过日志和事件...
8. **DAGScheduler和TaskScheduler**:Spark的调度系统,DAGScheduler将用户作业转化为Stage,而TaskScheduler负责将Stage分解为任务并在Executor上执行。 9. **YARN和Mesos集成**:Spark可以运行在Hadoop的YARN或...
3. **调度器(Scheduler)**:Spark的调度器分为两层,首先是DAGScheduler,负责将DAG拆分成Stage,然后是TaskScheduler,负责将Task分配给Executor。 4. **存储系统(Storage)**:Spark的存储模块管理RDD的持久化...
在 Spark 调度中最重要的是 DAGScheduler 和 TaskScheduler 两个调度器,其中,DAGScheduler 负责任务的逻辑调度, 将作业拆分为不同阶段的具有依赖关系的任务集。TaskScheduler 则负责具体任务的调度执行...
2. **任务类型支持**:DolphinScheduler支持多种任务类型,包括Shell、Java、Python、Spark、Hadoop等,覆盖了大数据处理的常见需求。 3. **资源管理**:系统允许用户定义和管理资源,如HDFS路径、数据库连接等,...
如果要用到MySQL数据,需要将pom.xml中MySQL的scope注释掉 当前只测试了MySQL、PostgreSQL和HIVE数据源,其他数据源暂时未测试过 Spark需要配置好读取Hive元数据,Spark不是采用jdbc的方式读取Hive
- 在Web界面中,用户可以创建项目,设计工作流,添加任务节点(如SQL、Shell、Spark等),定义任务间的依赖关系。 - 调度策略可以在工作流定义中设置,如按时间间隔、上一个任务完成等条件触发。 - 监控页面可...
在`apache-dolphinscheduler-3.2.0-bin`压缩包中,我们可以找到以下关键组件: 1. `bin`目录:包含启动和停止DolphinScheduler服务的脚本,如`start.sh`和`stop.sh`,以及用于管理服务的其他命令行工具。 2. `conf...
在Spark中,调度器(Scheduler)是整个系统的核心组件之一,负责管理和调度任务(task)以及执行过程中的各类资源分配。本文档将深入探讨Spark调度器的原理,从其架构设计到实际的调度过程,帮助读者对Spark的调度...
首先,Spark的核心是它的弹性分布式数据集(Resilient Distributed Datasets, RDD),这是一种不可变的数据结构,可以在集群中进行并行操作。RDD的分区分布在多台机器上,提供容错能力,使得在节点故障时能够自动...
**dolphinScheduler**(简称DS)是一款强大的分布式任务调度平台,支持多种类型的作业执行,如Shell、Python、Spark等。它能够高效地管理任务间的依赖关系,并提供了丰富的功能来满足企业级的数据处理需求。在大数据...