标签：des style blog class code java
首先，来看怎么构造一个org.apache.hadoop.yarn.client.api.YarnClient
|
1
2
3
4
5 |
class ClientHelper(conf: Configuration) extends
Logging { val yarnClient = YarnClient.createYarnClient info("trying to connect to RM %s"
format conf.get(YarnConfiguration.RM_ADDRESS, YarnConfiguration.DEFAULT_RM_ADDRESS)) yarnClient.init(conf); yarnClient.start |
！！！这个client还有个start方法，看来它跟RM很谈得来。的确，它实现了service这个接口。 好吧，它是一个服务。在YarnJobFactory中，我们用yarn-site.xml构造了一个YarnConfiguration对象，现在用它来初始化YarnClient，因为我们至少需要RM在哪，对不？
下边分几部分看submitApplication方法的实现
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 |
def submitApplication(packagePath: Path, memoryMb: Int, cpuCore: Int, cmds: List[String], env: Option[Map[String, String]], name: Option[String]): Option[ApplicationId] = { val app = yarnClient.createApplication val newAppResponse = app.getNewApplicationResponse var mem = memoryMb var cpu = cpuCore // If we are asking for memory more than the max allowed, shout out if
(mem > newAppResponse.getMaximumResourceCapability().getMemory()) { throw
new SamzaException("You‘re asking for more memory (%s) than is allowed by YARN: %s"
format (mem, newAppResponse.getMaximumResourceCapability().getMemory())) } // If we are asking for cpu more than the max allowed, shout out if
(cpu > newAppResponse.getMaximumResourceCapability().getVirtualCores()) { throw
new SamzaException("You‘re asking for more CPU (%s) than is allowed by YARN: %s"
format (cpu, newAppResponse.getMaximumResourceCapability().getVirtualCores())) } appId = Some(newAppResponse.getApplicationId) |
首先通过yarnClient的createApplication方法获取一个YarnClientApplication对象。这是对RM的第一次请求，那么这次请求能得到什么信息呢？
通过这次请求得到的YarnClientApplication对象有两个方法：
ApplicationSubmissionContext represents
all of the information needed by
the ResourceManager to launch
the ApplicationMaster for an
application.”鉴于YarnClient的createApplication方法没有任何参数，而YarnClient本身的状态中由用户指定的部分只是YarnConfiguration的内容，因此这个createApplication方法并不会告诉YARN客户端对资源的需求，因此它返回的app对象只包含了yarn的RM本身的信息。
在获取了app这个对象之后，submitApplication方法通过
|
1 |
val newAppResponse = app.getNewApplicationResponse |
从中取出了newAppResponse这个对象，然后从中取出了当前YARN集群最多支持的内存和CPU数目(TODO:这个值是当前可用的资源的值，还是整体上最大资源值)。然后对比给AM申请的container想要的内存和CPU，如果超出了YARN支持的最大值，就抛出异常。
否则，就把从newAppResponse中获取的applicationId赋给appId。看来在第一次请求时，YARN就给分配了appId，只是这个appId，并不和资源关联。
如果资源足够，AM就可以提交，那就开始填写AM运行需要的资源，具体来说就是组装ApplicationSubmissionContext类的一个对象
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39 |
name match { case
Some(name) => { appCtx.setApplicationName(name) } case
None => { appCtx.setApplicationName(appId.toString) }}env match { case
Some(env) => { containerCtx.setEnvironment(env) info("set environment variables to %s for %s"
format (env, appId.get)) } case
None => None}// set the local package so that the containers and app master are provisioned with itval packageUrl = ConverterUtils.getYarnUrlFromPath(packagePath)val fileStatus = packagePath.getFileSystem(conf).getFileStatus(packagePath)packageResource.setResource(packageUrl)info("set package url to %s for %s"
format (packageUrl, appId.get))packageResource.setSize(fileStatus.getLen)info("set package size to %s for %s"
format (fileStatus.getLen, appId.get))packageResource.setTimestamp(fileStatus.getModificationTime)packageResource.setType(LocalResourceType.ARCHIVE)packageResource.setVisibility(LocalResourceVisibility.APPLICATION)resource.setMemory(mem)info("set memory request to %s for %s"
format (mem, appId.get))resource.setVirtualCores(cpu)info("set cpu core request to %s for %s"
format (cpu, appId.get))appCtx.setResource(resource)containerCtx.setCommands(cmds.toList)info("set command to %s for %s"
format (cmds, appId.get))containerCtx.setLocalResources(Collections.singletonMap("__package", packageResource))appCtx.setApplicationId(appId.get)info("set app ID to %s"
format appId.get)appCtx.setAMContainerSpec(containerCtx)appCtx.setApplicationType(ClientHelper.applicationType)info("submitting application request for %s"
format appId.get)yarnClient.submitApplication(appCtx) |
这段代码设置了一个ApplicationSubmissionContext对象，然后再用yarnClient把它提交。这样就提交了一个YARN job。
这样YarnClient一共用了两次，初始一次请求，获取appID和YARN的资源上限的情况，第二次请求，真正提交job。
这段代码让我有些疑惑。首先appCtx大致分为两部分，一部分是job的信息，比如application type和application ID，另一部分和AM有关。和AM有关的部分又可以分成两块： 1. cpu和内存的大小，这两个资源组装在Resource这个类的对象里，由setResource设置到 appCtx中 2：运行container所需的命令和文件、环量变量，这部分设置在一个ContainerLaunchContext对象中，然后这个对象再被调置在appCtx中。疑惑的地方在于：为什么AM所需的资源要分成两部分呢？cpu和内存本就该是container申请的一部分呀？
看看API里关于containerLaunchContext类的说明，就更不明白了
ContainerLaunchContextrepresents all of the information needed by theNodeManagerto launch a container.It includes details such as:
ContainerIdof the container.Resourceallocated to the container.- User to whom the container is allocated.
- Security tokens (if security is enabled).
LocalResourcenecessary for running the container such as binaries, jar, shared-objects, side-files etc.- Optional, application-specific binary service data.
- Environment variables for the launched process.
- Command to launch the container.
好吧，“Resource allocated
to the container.”,
这一条ContainerLanchContext并没有体现，在它提供的方法中并不能设置Resource。这不是骗人吗？
而appCtx却有单独的一个setAMContainerSpec 方法来设置Resource。那么在申请运行task所需的container时，如果说明其所需的资源呢？看来一定不是用了这个ContainerLaunchContext对象。
Samza AM为task申请container的代码在SamzaAppMasterTaskManager这个类里
|
1
2
3
4
5
6
7
8
9 |
protected
def requestContainers(memMb: Int, cpuCores: Int, containers: Int) { info("Requesting %d container(s) with %dmb of memory"
format (containers, memMb)) val capability = Records.newRecord(classOf[Resource]) val priority = Records.newRecord(classOf[Priority]) priority.setPriority(0) capability.setMemory(memMb) capability.setVirtualCores(cpuCores) (0
until containers).foreach(idx => amClient.addContainerRequest(new
ContainerRequest(capability, null, null, priority)))} |
这里的amClient就是org.apache.hadoop.yarn.client.api.async.AMRMClientAsync类的对象。它用来和RM联系，处理container相关的事情。当AM请求container时，它就不用submitApplication中为AM设置container资源所需的那套动作了，而是使用ContainerRequest这类。而且ContainerRequest的构造方法中
|
1 |
public ContainerRequest(Resource capability, String[] nodes, String[] racks, Priority priority, boolean
relaxLocality) |
使用了Resource做为参数。
可见为AM申请container和为task申请container走的过程的确不一样。毕竟，为AM的运行申请container是作为提交任务的一部分。最终发现两个是使用的不同的协议。提交任务时，使用的是这个协议：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 |
message ApplicationSubmissionContextProto { optional ApplicationIdProto application_id = 1; optional string application_name = 2
[default
= "N/A"]; optional string queue = 3
[default
= "default"]; optional PriorityProto priority = 4; optional ContainerLaunchContextProto am_container_spec = 5; optional bool cancel_tokens_when_complete = 6
[default
= true]; optional bool unmanaged_am = 7
[default
= false]; optional int32 maxAppAttempts = 8
[default
= 0]; optional ResourceProto resource = 9; optional string applicationType = 10
[default
= "YARN"];}message ContainerLaunchContextProto { repeated StringLocalResourceMapProto localResources = 1; optional bytes tokens = 2; repeated StringBytesMapProto service_data = 3; repeated StringStringMapProto environment = 4; repeated string command = 5; repeated ApplicationACLMapProto application_ACLs = 6;} |
ContainerLaunchContextProto里根本没有代表cpu和内存资源的ResourceProto，这个Protocol是在ApplicationSubmissionContextProto里。对照containerLaunchContext类的说明，的确显得很奇怪。
而申请container的请求，走的是
|
1
2
3
4
5
6
7
8
9
10
11
12 |
message ResourceRequestProto { optional PriorityProto priority = 1; optional string resource_name = 2; optional ResourceProto capability = 3; optional int32 num_containers = 4; optional bool relax_locality = 5
[default
= true];}message ResourceProto { optional int32 memory = 1; optional int32 virtual_cores = 2;} |
Samza在YARN上的启动过程 =》 之二 submitApplication,布布扣,bubuko.com
Samza在YARN上的启动过程 =》 之二 submitApplication
标签：des style blog class code java
原文地址：http://www.cnblogs.com/devos/p/3720174.html
