hadoop - yarn application

**bordi** · 08/09/2015, 21h52

J'ai pas mal joué avec le map reduce, maintenant je découvre la partie yarn application,

fonctionnalité d'hadoop 2 qui permet d'étendre les capacités de distribution du client sur les nodes du cluster

j'ai trouvé un exemple sympa que j'ai bricolé et adapter pour hadoop 2.6.0

On commence par le client, il va d'abord chercher à se connecter en mode client et demander
la creation d'une application à yarn puis on va récupérer le contexte qui va nous permettre
de définir les ressources de notre jar qu'on souhaite executer sur les nodes.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
 ContainerLaunchContext localContainerLaunchContext = (ContainerLaunchContext)Records.newRecord(ContainerLaunchContext.class);
    localContainerLaunchContext.setCommands(Collections.singletonList("$JAVA_HOME/bin/java -Xmx256M yarnhello.ApplicationMaster 1><LOG_DIR>/stdout 2><LOG_DIR>/stderr"));

une fois qu'on aura fait tout ca on demandera à publier sur les nodes du cluster.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
ApplicationId localApplicationId = localApplicationSubmissionContext.getApplicationId();
    System.out.println("Client: Submitting " + localApplicationId);
    localYarnClient.submitApplication(localApplicationSubmissionContext);

Client complet

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
package yarnhello;


import java.io.PrintStream;
import java.util.Collections;
import java.util.HashMap;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.Records;

public class Client
{
  private YarnConfiguration conf;

  public static void main(String[] paramArrayOfString)
  {
    System.out.println("Client: Initializing");
    try {
      new Client().run();
    } catch (Exception localException) {
      localException.printStackTrace();
    }
  }

  private void run() throws Exception {
    this.conf = new YarnConfiguration();

    YarnClient localYarnClient = YarnClient.createYarnClient();
    localYarnClient.init(this.conf);
    localYarnClient.start();

    YarnClientApplication localYarnClientApplication = localYarnClient.createApplication();

    ContainerLaunchContext localContainerLaunchContext = (ContainerLaunchContext)Records.newRecord(ContainerLaunchContext.class);
    localContainerLaunchContext.setCommands(Collections.singletonList("$JAVA_HOME/bin/java -Xmx256M yarnhello.ApplicationMaster 1><LOG_DIR>/stdout 2><LOG_DIR>/stderr"));

    LocalResource localLocalResource = (LocalResource)Records.newRecord(LocalResource.class);
    Utils.setUpLocalResource(Utils.YARNAPP_JAR_PATH, localLocalResource, this.conf);
    localContainerLaunchContext.setLocalResources(Collections.singletonMap("YARNHELLO.jar", localLocalResource));

    HashMap localHashMap = new HashMap();
    Utils.setUpEnv(localHashMap, this.conf);
    
    localContainerLaunchContext.setEnvironment(localHashMap);

    Resource localResource = (Resource)Records.newRecord(Resource.class);
    localResource.setMemory(256);
    localResource.setVirtualCores(1);

    ApplicationSubmissionContext localApplicationSubmissionContext = localYarnClientApplication.getApplicationSubmissionContext();
    localApplicationSubmissionContext.setApplicationName("YARNHELLO");
    localApplicationSubmissionContext.setQueue("default");
    localApplicationSubmissionContext.setAMContainerSpec(localContainerLaunchContext);
    localApplicationSubmissionContext.setResource(localResource);

    ApplicationId localApplicationId = localApplicationSubmissionContext.getApplicationId();
    System.out.println("Client: Submitting " + localApplicationId);
    localYarnClient.submitApplication(localApplicationSubmissionContext);

    ApplicationReport localApplicationReport = localYarnClient.getApplicationReport(localApplicationId);
    YarnApplicationState localYarnApplicationState = localApplicationReport.getYarnApplicationState();

    while ((localYarnApplicationState != YarnApplicationState.FINISHED) && (localYarnApplicationState != YarnApplicationState.KILLED) && (localYarnApplicationState != YarnApplicationState.FAILED)) {
      Thread.sleep(1000L);
      localApplicationReport = localYarnClient.getApplicationReport(localApplicationId);
      localYarnApplicationState = localApplicationReport.getYarnApplicationState();
    }

    System.out.println("Client: Finished " + localApplicationId + " with state " + localYarnApplicationState);
  }
}

ApplicationMaster, va être déployer sur les nodes selon la limite définie (containerCount = 3), puis il va lancer 3 fois le conteneur contenant ma classe Coucou

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
 ContainerLaunchContext localContainerLaunchContext = (ContainerLaunchContext)Records.newRecord(ContainerLaunchContext.class);
      localContainerLaunchContext.setCommands(Collections.singletonList("$JAVA_HOME/bin/java -Xmx256M yarnhello.Coucou 1><LOG_DIR>/stdout 2><LOG_DIR>/stderr"));

this.nmClient.startContainer(localContainer,localContainerLaunchContext )

l'application master va se deployer sur les noeuds et se prepare à lancer mes conteneurs avec mon jar java
il va se connecter au RessourceManager et au NodeManager

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
package yarnhello;


import java.io.PrintStream;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;

import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.NMClient;
import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync.CallbackHandler;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.Records;

public class ApplicationMaster
  implements AMRMClientAsync.CallbackHandler
{
  private YarnConfiguration conf = new YarnConfiguration();
  private NMClient nmClient;
  private int containerCount = 3;

  public static void main(String[] paramArrayOfString) {
    System.out.println("AppMaster: Initializing");
    try {
      new ApplicationMaster().run();
    } catch (Exception localException) {
      localException.printStackTrace();
    }
  }

  public void run() throws Exception {
    this.conf = new YarnConfiguration();

    this.nmClient = NMClient.createNMClient();
    this.nmClient.init(this.conf);
    this.nmClient.start();

    AMRMClientAsync localAMRMClientAsync = AMRMClientAsync.createAMRMClientAsync(1000, this);
    localAMRMClientAsync.init(this.conf);
    localAMRMClientAsync.start();

    localAMRMClientAsync.registerApplicationMaster("", 0, "");
    System.out.println("AppMaster: Registered");

    Priority localPriority = (Priority)Records.newRecord(Priority.class);
    localPriority.setPriority(0);

    Resource localResource = (Resource)Records.newRecord(Resource.class);
    localResource.setMemory(128);
    localResource.setVirtualCores(1);

    System.out.println("AppMaster: Requesting " + this.containerCount + " Containers");
    for (int i = 0; i < this.containerCount; i++) {
      localAMRMClientAsync.addContainerRequest(new AMRMClient.ContainerRequest(localResource, null, null, localPriority));
    }

    while (!containersFinished()) {
      Thread.sleep(100L);
    }

    System.out.println("AppMaster: Unregistered");
    localAMRMClientAsync.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED, "", "");
  }

  private boolean containersFinished() {
    return this.containerCount == 0;
  }

  public void onContainersAllocated(List<Container> paramList)
  {
    for (Container localContainer : paramList)
      try {
        this.nmClient.startContainer(localContainer, initContainer());
        System.err.println("AppMaster: Container launched " + localContainer.getId());
      } catch (Exception localException) {
        System.err.println("AppMaster: Container not launched " + localContainer.getId());
        localException.printStackTrace();
      }
  }

  private ContainerLaunchContext initContainer()
  {
    try
    {
      ContainerLaunchContext localContainerLaunchContext = (ContainerLaunchContext)Records.newRecord(ContainerLaunchContext.class);
      localContainerLaunchContext.setCommands(Collections.singletonList("$JAVA_HOME/bin/java -Xmx256M yarnhello.Coucou 1><LOG_DIR>/stdout 2><LOG_DIR>/stderr"));

      LocalResource localLocalResource = (LocalResource)Records.newRecord(LocalResource.class);
      Utils.setUpLocalResource(Utils.YARNAPP_JAR_PATH, localLocalResource, this.conf);
      localContainerLaunchContext.setLocalResources(Collections.singletonMap("YARNHELLO.jar", localLocalResource));

      HashMap localHashMap = new HashMap();
      Utils.setUpEnv(localHashMap, this.conf);
      localContainerLaunchContext.setEnvironment(localHashMap);

      return localContainerLaunchContext;
    } catch (Exception localException) {
      localException.printStackTrace();
    }return null;
  }

  public void onContainersCompleted(List<ContainerStatus> paramList)
  {
    for (ContainerStatus localContainerStatus : paramList) {
      System.err.println("AppMaster: Container finished " + localContainerStatus.getContainerId());
      synchronized (this) {
        this.containerCount -= 1;
      }
    }
  }

  public void onError(Throwable paramThrowable)
  {
  }

  public void onNodesUpdated(List<NodeReport> paramList) {
  }

  public void onShutdownRequest() {
  }

  public float getProgress() {
    return 0.0F;
  }
}

Ma classe bidon Coucou qui peut contenir tout ce que je veux

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
package yarnhello;

import java.io.PrintStream;

public class Coucou
{
  public static void main(String[] paramArrayOfString)
    throws Exception
  {
    System.out.println("Initialisation du conteneur");
    System.out.println("Salut tout le monde ! bienvenu dans la boite à coucou");
    System.out.println("Fin d'execution du conteneur");
  }
}

edit: Classe util, ca peut servir

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
40
41
42
43
44
45
46
47
48
49
50
51
package yarnhello;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.ConverterUtils;

public class Utils
{
  public static final String YARNAPP_JAR_NAME = "YARNHELLO.jar";
  public static final Path YARNAPP_JAR_PATH = new Path("/apps/YARNHELLO.jar");

  public static void setUpEnv(Map<String, String> paramMap, YarnConfiguration paramYarnConfiguration) {
    StringBuilder localStringBuilder = new StringBuilder(ApplicationConstants.Environment.CLASSPATH.$$()).append("<CPS>").append("./*");

    for (String str : paramYarnConfiguration.getStrings("yarn.application.classpath", YarnConfiguration.DEFAULT_YARN_CROSS_PLATFORM_APPLICATION_CLASSPATH))
    {
      localStringBuilder.append("<CPS>");
      localStringBuilder.append(str.trim());
    }
    localStringBuilder.append("<CPS>").append("./log4j.properties");

    if (paramYarnConfiguration.getBoolean("yarn.is.minicluster", false)) {
      localStringBuilder.append(':');
      localStringBuilder.append(System.getProperty("java.class.path"));
    }
    paramMap.put("CLASSPATH", localStringBuilder.toString());
  }

  public static void setUpLocalResource(Path paramPath, LocalResource paramLocalResource, YarnConfiguration paramYarnConfiguration) throws IOException {
    Path localPath = FileContext.getFileContext().makeQualified(paramPath);

    FileStatus localFileStatus = FileSystem.get(paramYarnConfiguration).getFileStatus(localPath);
    paramLocalResource.setResource(ConverterUtils.getYarnUrlFromPath(localPath));
    paramLocalResource.setSize(localFileStatus.getLen());
    paramLocalResource.setTimestamp(localFileStatus.getModificationTime());
    paramLocalResource.setType(LocalResourceType.FILE);
    paramLocalResource.setVisibility(LocalResourceVisibility.PUBLIC);
  }
}

Donc il lance une premiere fois mon conteneur sur mon node itx01, puis deux autres mon autre
node itx02

15/09/08 20:36:41 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15/09/08 20:36:42 INFO fs.TrashPolicyDefault: Namenode trash configuration: Deletion interval = 0 minutes, Emptier interval = 0 minutes.
Deleted /apps/YARNHELLO.jar
hduser@stargate:~$ hadoop fs -copyFromLocal YARNHELLO.jar /apps
15/09/08 20:36:59 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
hduser@stargate:~$ hadoop jar YARNHELLO.jar yarnhello.Client
Client: Initializing
15/09/08 20:37:13 INFO client.RMProxy: Connecting to ResourceManager at stargate/192.168.0.11:8032
15/09/08 20:37:13 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Client: Submitting application_1441649620937_0009
15/09/08 20:37:14 INFO impl.YarnClientImpl: Submitted application application_1441649620937_0009
Client: Finished application_1441649620937_0009 with state FINISHED
hduser@stargate:~$ yarn logs -applicationId application_1441649620937_0009
15/09/08 20:37:59 INFO client.RMProxy: Connecting to ResourceManager at stargate/192.168.0.11:8032
15/09/08 20:37:59 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

Container: container_1441649620937_0009_01_000003 on itx01_38328
==================================================================
LogType:stderr
Log Upload Time:8-sept.-2015 20:37:28
LogLength:0
Log Contents:

LogType:stdout
Log Upload Time:8-sept.-2015 20:37:28
LogLength:112
Log Contents:
Initialisation du conteneur
Salut tout le monde ! bienvenu dans la boite Ã* coucou
Fin d'execution du conteneur

Container: container_1441649620937_0009_01_000004 on itx02_47715
==================================================================
LogType:stderr
Log Upload Time:8-sept.-2015 20:37:27
LogLength:0
Log Contents:

LogType:stdout
Log Upload Time:8-sept.-2015 20:37:27
LogLength:112
Log Contents:
Initialisation du conteneur
Salut tout le monde ! bienvenu dans la boite Ã* coucou
Fin d'execution du conteneur

Container: container_1441649620937_0009_01_000001 on itx02_47715
==================================================================
LogType:stderr
Log Upload Time:8-sept.-2015 20:37:27
LogLength:1391
Log Contents:
15/09/08 20:37:31 INFO impl.ContainerManagementProtocolProxy: yarn.client.max-cached-nodemanagers-proxies : 0
15/09/08 20:37:32 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15/09/08 20:37:32 INFO client.RMProxy: Connecting to ResourceManager at stargate/192.168.0.11:8030
15/09/08 20:37:36 INFO impl.AMRMClientImpl: Received new token for : stargate:42554
15/09/08 20:37:36 INFO impl.AMRMClientImpl: Received new token for : itx01:38328
15/09/08 20:37:36 INFO impl.AMRMClientImpl: Received new token for : itx02:47715
15/09/08 20:37:38 INFO impl.ContainerManagementProtocolProxy: Opening proxy : stargate:42554
AppMaster: Container launched container_1441649620937_0009_01_000002
15/09/08 20:37:38 INFO impl.ContainerManagementProtocolProxy: Opening proxy : itx01:38328
AppMaster: Container launched container_1441649620937_0009_01_000003
15/09/08 20:37:39 INFO impl.ContainerManagementProtocolProxy: Opening proxy : itx02:47715
AppMaster: Container launched container_1441649620937_0009_01_000004
AppMaster: Container finished container_1441649620937_0009_01_000002
AppMaster: Container finished container_1441649620937_0009_01_000003
AppMaster: Container finished container_1441649620937_0009_01_000004
15/09/08 20:37:40 INFO impl.AMRMClientImpl: Waiting for application to be successfully unregistered.

LogType:stdout
Log Upload Time:8-sept.-2015 20:37:27
LogLength:105
Log Contents:
AppMaster: Initializing
AppMaster: Registered
AppMaster: Requesting 3 Containers
AppMaster: Unregistered

Container: container_1441649620937_0009_01_000002 on stargate_42554
=====================================================================
LogType:stderr
Log Upload Time:8-sept.-2015 20:37:28
LogLength:0
Log Contents:

LogType:stdout
Log Upload Time:8-sept.-2015 20:37:28
LogLength:112
Log Contents:
Initialisation du conteneur
Salut tout le monde ! bienvenu dans la boite Ã* coucou
Fin d'execution du conteneur

Je suis en train de décortiquer l'exemple d'executeur de commande shell distribué d'hortonworks, il est un peu plus balaise.

**bordi** · 08/09/2015, 22h07

Autre exemple, il vient d'hortonworks, qui doit être une version simplifié de celui que j'analyse.

simple-yarn-app

Le client qui prepare et lance l'application master après s'être connecter, il prend en compte
les arguments que je lui ai passé, la commande shlle /usr/bin/calendar à executer, le nombre d'instance,
l'emplacement du jar sur l'hdfs.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
package com.hortonworks.simpleyarnapp;

import java.io.File;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.Apps;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;


public class Client {

  Configuration conf = new YarnConfiguration();
  
  public void run(String[] args) throws Exception {
    final String command = args[0];
    final int n = Integer.valueOf(args[1]);
    final Path jarPath = new Path(args[2]);

    // Create yarnClient
    YarnConfiguration conf = new YarnConfiguration();
    YarnClient yarnClient = YarnClient.createYarnClient();
    yarnClient.init(conf);
    yarnClient.start();
    
    // Create application via yarnClient
    YarnClientApplication app = yarnClient.createApplication();

    // Set up the container launch context for the application master
    ContainerLaunchContext amContainer = 
        Records.newRecord(ContainerLaunchContext.class);
    amContainer.setCommands(
        Collections.singletonList(
            "$JAVA_HOME/bin/java" +
            " -Xmx256M" +
            " com.hortonworks.simpleyarnapp.ApplicationMaster" +
            " " + command +
            " " + String.valueOf(n) +
            " 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout" + 
            " 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr" 
            )
        );
    
    // Setup jar for ApplicationMaster
    LocalResource appMasterJar = Records.newRecord(LocalResource.class);
    setupAppMasterJar(jarPath, appMasterJar);
    amContainer.setLocalResources(
        Collections.singletonMap("simpleapp.jar", appMasterJar));

    // Setup CLASSPATH for ApplicationMaster
    Map<String, String> appMasterEnv = new HashMap<String, String>();
    setupAppMasterEnv(appMasterEnv);
    amContainer.setEnvironment(appMasterEnv);
    
    // Set up resource type requirements for ApplicationMaster
    Resource capability = Records.newRecord(Resource.class);
    capability.setMemory(256);
    capability.setVirtualCores(1);

    // Finally, set-up ApplicationSubmissionContext for the application
    ApplicationSubmissionContext appContext = 
    app.getApplicationSubmissionContext();
    appContext.setApplicationName("simple-yarn-app"); // application name
    appContext.setAMContainerSpec(amContainer);
    appContext.setResource(capability);
    appContext.setQueue("default"); // queue 

    // Submit application
    ApplicationId appId = appContext.getApplicationId();
    System.out.println("Submitting application " + appId);
    yarnClient.submitApplication(appContext);
    
    ApplicationReport appReport = yarnClient.getApplicationReport(appId);
    YarnApplicationState appState = appReport.getYarnApplicationState();
    while (appState != YarnApplicationState.FINISHED && 
           appState != YarnApplicationState.KILLED && 
           appState != YarnApplicationState.FAILED) {
      Thread.sleep(100);
      appReport = yarnClient.getApplicationReport(appId);
      appState = appReport.getYarnApplicationState();
    }
    
    System.out.println(
        "Application " + appId + " finished with" +
    		" state " + appState + 
    		" at " + appReport.getFinishTime());

  }
  
  private void setupAppMasterJar(Path jarPath, LocalResource appMasterJar) throws IOException {
    FileStatus jarStat = FileSystem.get(conf).getFileStatus(jarPath);
    appMasterJar.setResource(ConverterUtils.getYarnUrlFromPath(jarPath));
    appMasterJar.setSize(jarStat.getLen());
    appMasterJar.setTimestamp(jarStat.getModificationTime());
    appMasterJar.setType(LocalResourceType.FILE);
    appMasterJar.setVisibility(LocalResourceVisibility.PUBLIC);
  }
  
  private void setupAppMasterEnv(Map<String, String> appMasterEnv) {
    for (String c : conf.getStrings(
        YarnConfiguration.YARN_APPLICATION_CLASSPATH,
        YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH)) {
      Apps.addToEnvironment(appMasterEnv, Environment.CLASSPATH.name(),
          c.trim());
    }
    Apps.addToEnvironment(appMasterEnv,
        Environment.CLASSPATH.name(),
        Environment.PWD.$() + File.separator + "*");
  }
  
  public static void main(String[] args) throws Exception {
    Client c = new Client();
    c.run(args);
  }
}

L'application master qui prend en argument de commande /usr/bin/calendar
pour l'exectuer au niveau des nodes, ainsi que le nombre d'instance conteneur à gerer

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
package com.hortonworks.simpleyarnapp;


import java.util.Collections;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.NMClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.Records;

public class ApplicationMaster {

  public static void main(String[] args) throws Exception {

    final String command = args[0];
    final int n = Integer.valueOf(args[1]);
    
    // Initialize clients to ResourceManager and NodeManagers
    Configuration conf = new YarnConfiguration();

    AMRMClient<ContainerRequest> rmClient = AMRMClient.createAMRMClient();
    rmClient.init(conf);
    rmClient.start();

    NMClient nmClient = NMClient.createNMClient();
    nmClient.init(conf);
    nmClient.start();

    // Register with ResourceManager
    System.out.println("registerApplicationMaster 0");
    rmClient.registerApplicationMaster("", 0, "");
    System.out.println("registerApplicationMaster 1");
    
    // Priority for worker containers - priorities are intra-application
    Priority priority = Records.newRecord(Priority.class);
    priority.setPriority(0);

    // Resource requirements for worker containers
    Resource capability = Records.newRecord(Resource.class);
    capability.setMemory(128);
    capability.setVirtualCores(1);

    // Make container requests to ResourceManager
    for (int i = 0; i < n; ++i) {
      ContainerRequest containerAsk = new ContainerRequest(capability, null, null, priority);
      System.out.println("Making res-req " + i);
      rmClient.addContainerRequest(containerAsk);
    }

    // Obtain allocated containers, launch and check for responses
    int responseId = 0;
    int completedContainers = 0;
    while (completedContainers < n) {
        AllocateResponse response = rmClient.allocate(responseId++);
        for (Container container : response.getAllocatedContainers()) {
            // Launch container by create ContainerLaunchContext
            ContainerLaunchContext ctx =
                    Records.newRecord(ContainerLaunchContext.class);
            ctx.setCommands(
                    Collections.singletonList(
                            command +
                                    " 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout" +
                                    " 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr"
                    ));
            System.out.println("Launching container " + container.getId());
            nmClient.startContainer(container, ctx);
        }
        for (ContainerStatus status : response.getCompletedContainersStatuses()) {
            ++completedContainers;
            System.out.println("Completed container " + status.getContainerId());
        }
        Thread.sleep(100);
    }

    // Un-register with ResourceManager
    rmClient.unregisterApplicationMaster(
        FinalApplicationStatus.SUCCEEDED, "", "");
  }
}

logs

hadoop jar simple-yarn-app-1.1.0.jar com.hortonworks.simpleyarnapp.Client /usr/bin/calendar 2 hdfs:///apps/simple/simple-yarn-app-1.1.0.jar
15/09/08 21:11:08 INFO client.RMProxy: Connecting to ResourceManager at stargate/192.168.0.11:8032
15/09/08 21:11:08 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Submitting application application_1441649620937_0010
15/09/08 21:11:10 INFO impl.YarnClientImpl: Submitted application application_1441649620937_0010
Application application_1441649620937_0010 finished with state FINISHED at 1441739476704

yarn logs -applicationId application_1441649620937_0010
15/09/08 21:11:47 INFO client.RMProxy: Connecting to ResourceManager at stargate/192.168.0.11:8032
15/09/08 21:11:47 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

Container: container_1441649620937_0010_01_000002 on itx01_38328
==================================================================
LogType:stderr
Log Upload Time:8-sept.-2015 21:11:17
LogLength:0
Log Contents:

LogType:stdout
Log Upload Time:8-sept.-2015 21:11:17
LogLength:1466
Log Contents:
sept. 08 Richard ``the Lionheart'', king of England born in Oxford, 1157
sept. 08 Peter Sellers born in Southsea, England, 1925
sept. 08 "Star Trek" debuts on NBC (1966)
sept. 08 Jack the Ripper kills again, Annie Chapman is second victim, 1888
sept. 08 US President Ford pardons Richard M. Nixon, 1974
sept. 08 Antonin Dvorak born in Nelahozeves, Bohemia, 1841
sept. 08 Richard Strauss dies in Garmisch-Partenkirchen, Germany, 1949
sept. 08 Ron "Pigpen" McKernan (Grateful Dead) is born in San Bruno, California, 1945
sept. 08 DÃ*a del agrigultor
sept. 08 Boris Samorodov <bsam@FreeBSD.org> born in Krasnodar, Russian Federation, 1963
sept. 08 Richard Strauss in Garmisch-Partenkirchen gestorben, 1949
sept. 08 MÃ¡ria, Adrienn
sept. 09 Chinese Communist Party Chairman Mao Tse-Tung dies at age 82, 1976
sept. 09 Dennis MacAlistair Ritchie, creator of C, born, 1941
sept. 09 First bug found, a moth in Harvard Mark II, 1947
sept. 09 California becomes the 31st state of the USA, 1850
sept. 09 United Colonies is renamed the United States, 1776
sept. 09 Admission Day in California
sept. 09 National Day in North Korea
sept. 09 Anniversary of the Socialist Revolution (2 days) in Bulgaria
sept. 09 Muere en La Plata el polÃ*tico radical Ricardo BalbÃ*n, 1981
sept. 09 Yoshio Mita <mita@FreeBSD.org> born in Hiroshima, Japan, 1972
sept. 09 N'oubliez pas les Alain !
sept. 09 Mao Tse-Tung gestorben im Alter von 82 Jahren, 1976
sept. 09 ÃdÃ¡m

Container: container_1441649620937_0010_01_000001 on itx02_47715
==================================================================
LogType:stderr
Log Upload Time:8-sept.-2015 21:11:17
LogLength:806
Log Contents:
15/09/08 21:11:28 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15/09/08 21:11:28 INFO client.RMProxy: Connecting to ResourceManager at stargate/192.168.0.11:8030
15/09/08 21:11:28 INFO impl.ContainerManagementProtocolProxy: yarn.client.max-cached-nodemanagers-proxies : 0
15/09/08 21:11:29 INFO impl.AMRMClientImpl: Received new token for : itx01:38328
15/09/08 21:11:29 INFO impl.ContainerManagementProtocolProxy: Opening proxy : itx01:38328
15/09/08 21:11:30 INFO impl.AMRMClientImpl: Received new token for : stargate:42554
15/09/08 21:11:30 INFO impl.ContainerManagementProtocolProxy: Opening proxy : stargate:42554
15/09/08 21:11:31 INFO impl.AMRMClientImpl: Waiting for application to be successfully unregistered.

LogType:stdout
Log Upload Time:8-sept.-2015 21:11:17
LogLength:326
Log Contents:
registerApplicationMaster 0
registerApplicationMaster 1
Making res-req 0
Making res-req 1
Launching container container_1441649620937_0010_01_000002
Launching container container_1441649620937_0010_01_000003
Completed container container_1441649620937_0010_01_000003
Completed container container_1441649620937_0010_01_000002

Container: container_1441649620937_0010_01_000003 on stargate_42554
=====================================================================
LogType:stderr
Log Upload Time:8-sept.-2015 21:11:17
LogLength:0
Log Contents:

LogType:stdout
Log Upload Time:8-sept.-2015 21:11:17
LogLength:1466
Log Contents:
sept. 08 Richard ``the Lionheart'', king of England born in Oxford, 1157
sept. 08 Peter Sellers born in Southsea, England, 1925
sept. 08 "Star Trek" debuts on NBC (1966)
sept. 08 Jack the Ripper kills again, Annie Chapman is second victim, 1888
sept. 08 US President Ford pardons Richard M. Nixon, 1974
sept. 08 Antonin Dvorak born in Nelahozeves, Bohemia, 1841
sept. 08 Richard Strauss dies in Garmisch-Partenkirchen, Germany, 1949
sept. 08 Ron "Pigpen" McKernan (Grateful Dead) is born in San Bruno, California, 1945
sept. 08 DÃ*a del agrigultor
sept. 08 Boris Samorodov <bsam@FreeBSD.org> born in Krasnodar, Russian Federation, 1963
sept. 08 Richard Strauss in Garmisch-Partenkirchen gestorben, 1949
sept. 08 MÃ¡ria, Adrienn
sept. 09 Chinese Communist Party Chairman Mao Tse-Tung dies at age 82, 1976
sept. 09 Dennis MacAlistair Ritchie, creator of C, born, 1941
sept. 09 First bug found, a moth in Harvard Mark II, 1947
sept. 09 California becomes the 31st state of the USA, 1850
sept. 09 United Colonies is renamed the United States, 1776
sept. 09 Admission Day in California
sept. 09 National Day in North Korea
sept. 09 Anniversary of the Socialist Revolution (2 days) in Bulgaria
sept. 09 Muere en La Plata el polÃ*tico radical Ricardo BalbÃ*n, 1981
sept. 09 Yoshio Mita <mita@FreeBSD.org> born in Hiroshima, Japan, 1972
sept. 09 N'oubliez pas les Alain !
sept. 09 Mao Tse-Tung gestorben im Alter von 82 Jahren, 1976
sept. 09 ÃdÃ¡m

**bordi** · 08/09/2015, 22h28

Cette fonction permet de lancer n'importe quel type d'application dans le conteneur déployer par l'application master de yarn, c,python,scala,java,script,etc.. une extension non négligeable en plus du map reduce, mais gourmand en ressource, hbase et spark s'appui sur ce type de mécanisme.

hadoop - yarn application

Hadoop & co

Discussions similaires

Partager

Partager