Samza在YARN上的启动过程 =》 之一

运行脚本,提交job

往YARN提交Samza job要使用run-job.sh这个脚本。

samza-example/target/bin/run-job.sh  --config-factory=samza.config.factories.PropertiesConfigFactory  --config-path=file://$PWD/config/hello-world.properties

这脚本的内容是什么呢?

exec $(dirname $0)/run-class.sh org.apache.samza.job.JobRunner $@

它是调用run-class这个脚本。

run-class.sh会根据环境变量HADOOP_CONF_DIR和HADOOP_YARN_HOME获取YARN的配置文件位置,把它加入CLASSPATH。同时会把samza job根目录下的lib文件夹里的jar或war都加进CLASSPATH。只所以需要yarn的配置文件,目的是得到Resource Manager的地址。而lib目录下的包,是运行job必须的。

run-class.sh也会通过环境变量$SAMZA_LOG_DIR获知samza的log应存放的位置,通过$SAMZA_CONTAINER_NAME决定container的名字,然后把它们用-D设在JAVA_OPTS中。然后在lib目录下查找log4j.xml文件,存在的话,就把-Dlog4j.configuration设为log4j.xml的路径。

通过以上动作,构造好了调用java所需的classpath和任务运行时需要的一些配置项,然后调用

exec $JAVA $JAVA_OPTS -cp $CLASSPATH $@

启动虚拟机。

可以,在调用run-job.sh时,会运行org.apache.samza.job.JobRunner这个类。在samza的官方指南中简要介绍了一下这个类的作用。Samza自带了两种StreamJobFactory :LocalJobFactory 和 YarnJobFactory 。 StreamJobFactory的作用就是把给JobRunner提供一个可以执行的job

public interface StreamJobFactory {
StreamJob getJob(Config config);
}

而StreamJob就是一个可以执行的job, JobRunner会调用它的submit方法  

public interface StreamJob {
StreamJob submit(); StreamJob kill(); ApplicationStatus waitForFinish(long timeoutMs); ApplicationStatus waitForStatus(ApplicationStatus status, long timeoutMs); ApplicationStatus getStatus();
}

下边来看JobRunner这个类。程序入口在JobRunner这个类的伴生对象里

object JobRunner extends Logging {
def main(args: Array[String]) {
val cmdline = new CommandLine
val options = cmdline.parser.parse(args: _*)
val config = cmdline.loadConfig(options)
new JobRunner(config).run
}
}

  在配置参数以后,会走到JobRunner的run方法, 下边是它的主要逻辑

    val jobFactoryClass = conf.getStreamJobFactoryClass match {
case Some(factoryClass) => factoryClass
case _ => throw new SamzaException("no job factory class defined")
} val jobFactory = Class.forName(jobFactoryClass).newInstance.asInstanceOf[StreamJobFactory] // Create the actual job, and submit it.
val job = jobFactory.getJob(conf).submit //提交job info("waiting for job to start") // Wait until the job has started, then exit.
Option(job.waitForStatus(Running, 500)) match {
case Some(appStatus) => {
if (Running.equals(appStatus)) {
info("job started successfully")
} else {
warn("unable to start job successfully. job has status %s" format (appStatus))
}
}
case _ => warn("unable to start job successfully.")
}

  首先,它会去conf里找到是否设置了job.factory,即有没有指定StreamJobFactory的实现,没有就抛出异常退出。否则就通过这个StreamJobFactory提交job。在提交后,等待500毫秒,如果任务的状态不是Running就退出。这里的Running并不代表任务已经在跑了,比如在使用YARN时,只要成功提交给Resource Manager,就算是running了,所以这里的running是“任务提交成功“的意思。

YarnJob的实现

当提交给YARN时,我们使用YarnJobFactory这个StreamJobFactory的实现。

class YarnJobFactory extends StreamJobFactory {
def getJob(config: Config) = {
// TODO fix this. needed to support http package locations.
//这里会读yarn-site.xml。前提是yarn-site.xml必须在classpath里 在run-class.sh里,HADOOP_CONF_DIR路径被写进了classpath里
val hConfig = new YarnConfiguration
hConfig.set("fs.http.impl", classOf[HttpFileSystem].getName) new YarnJob(config, hConfig)
}
}

  

” hConfig.set("fs.http.impl", classOf[HttpFileSystem].getName)“ 这一句是使得可以在job里调用http文件系统,如把文件路径写成"http://xxx.xx.xx.xx:8080/xx/xx"这种。Samza自带了一个HTTP filsystem的实现。或许是LinkedIn的人需要这么用?

YarnJobFactory,主要就是构造了一个YarnConfiguration,和以前的commandLine参数一起作为config来构造一个YarnJob. YarnConfiguration是YARN自己的类,它会从classpath里读yarn-site.xml这个配置文件。

YarnJob这个才是要被提交给Yarn的任务,它实现了StreamJob这个接口。这里主要关心它的submit方法。

  val client = new ClientHelper(hadoopConfig)
  var appId: Option[ApplicationId] = None

 //提交job,注意会有AppMaster所需的内存和cpu数目。但不包括container数目
def submit: YarnJob = {
appId = client.submitApplication( //注意submitApplication的返回值是appId
new Path(config.getPackagePath.getOrElse(throw new SamzaException("No YARN package path defined in config."))),
config.getAMContainerMaxMemoryMb.getOrElse(DEFAULT_AM_CONTAINER_MEM),
1,
List(
"export SAMZA_LOG_DIR=%s && ln -sfn %s logs && exec ./__package/bin/run-am.sh 1>logs/%s 2>logs/%s"
format (ApplicationConstants.LOG_DIR_EXPANSION_VAR, ApplicationConstants.LOG_DIR_EXPANSION_VAR, ApplicationConstants.STDOUT, ApplicationConstants.STDERR)),
Some(Map(
ShellCommandConfig.ENV_CONFIG -> Util.envVarEscape(JsonConfigSerializer.toJson(config)),
ShellCommandConfig.ENV_CONTAINER_NAME -> Util.envVarEscape("application-master"),
ShellCommandConfig.ENV_JAVA_OPTS -> Util.envVarEscape(config.getAmOpts.getOrElse("")))),
Some("%s_%s" format (config.getName.get, config.getJobId.getOrElse(1)))) this
}

  submit方法来提交任务的过程交给了ClientHelper的submitAppliation来实现。

  这个方法才是提交YARN任务的关键。

  首先,我们知道要提交任务给YARN一定实现YARN指定的接口。那么来揣摩一下YARN需要我们提供给它什么东西,它才能调度一个job的执行。

  来猜猜看(实际上我已经知道了一些,看能不能想得更全一点,重要是理清思路)

  1. 首先,被提交的是什么?我已经知道YARN程序的执行过程是,先向YARN申请资源来运行一个 application master, 再由application master申请后续的资源。因此,这里提交的是一个请求,请求的内容是:RM,我要启动一个AM,给我分个container吧
  2. RM在什么时候才会同意一个application master的申请呢? 如果系统里的资源不足,有很多任务在运行,它会给拒了吗?
  3. 假如,RM批准了请求,给分了一个container。那么,那么……它只是一个container,毕竟没人使用它来执行程序。那么,谁来使用这个container来跑程序呢?那就是NodeManager了。那们在提交对AM的申请时,我们就需要告诉Yarn说我”需要NodeManager在这个container里这样……这样……这样做”
  4. 或许我们还该告诉Yarn这个AM需要的资源数量。毕竟运行这个AM也需要一个container,而一个container的核心就是它对应着一些资源。实际上,瞅一眼下边的方法签名,就会看到cpu和memory这两个大字。

  先看它的签名

   def submitApplication(packagePath: Path, memoryMb: Int, cpuCore: Int, cmds: List[String], env: Option[Map[String, String]], name: Option[String]): Option[ApplicationId]

下边介绍一下各个参数的含义,有助于我们了解这个方法都干了啥事。

  • packagePath 这个机制之前没有想到。我们实际上提交任务时,只需要告诉yarn这个任务所需要资源(文件资源)的地址,由NodeManager去down下来这个资源,做本地化。而不是把这个job package整个传输给yarn。这样设计使得获取任务资源更灵活,比如job package可以在http文件系统里,像Samza的例子里一样,也可以在HDFS里(这个得需要配一下,在Samza工程里所有以.md结尾里的文件里搜hdfs,就会找到使用HDFS存放job package的指南),当然也可以在本地(在没有成功搞在http和hdfs两种方式之前,曾经试过这样……)。好吧,这样起码不会给RM在存储上造成压力。
  • memoryMb 这个是想要给运行AM的container分配的内存大小
  • cpuCore 需要几个core,这个是虚拟的,具体怎么实现,有待研究……
  • cmds 这个就是NodeManage要执行命令吧
  • env 环境变量,这个是配给运行AM的虚拟机的
  • name 这个是这个job的名字,就是YARN job的名字,就是在RM的web UI上看到的名字。同时,Samza在bin目录下有个脚本,可以到当前在执行samza job。

返回值是ApplicationId。这是一个Option,所以提交失败时,返回值就是None。

再写下去submitApplication这个方法的实现,就有些太长,换下一篇

上一篇:android sdk manager 闪退 打不开问题


下一篇:02_Android写xml文件和读xml文件