src/main/java/org/apache/sysml/yarn/ropt/YarnClusterAnalyzer.java - systemds - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 package org.apache.sysml.yarn.ropt;

 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.StringTokenizer;

 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.yarn.api.records.NodeReport;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;

 import org.apache.sysml.hops.OptimizerUtils;
 import org.apache.sysml.runtime.matrix.mapred.MRConfigurationNames;

 /**
  * Central place for analyzing and obtaining static infrastructure properties
  * such as memory and number of logical processors.
  *
  *
  */
 public class YarnClusterAnalyzer
 {

 	public static final long DEFAULT_JVM_SIZE = 512 * 1024 * 1024;
 	public static final int CPU_HYPER_FACTOR = 1;

 	//static local master node properties
 	public static int  _localPar        = -1;
 	public static long _localJVMMaxMem  = -1;

 	//default hadoop cluster properties
 	public static int  _remotePar       = -1;
 	//public static int  _remoteParMap    = -1;
 	//public static int  _remoteParReduce = -1;
 	public static long _remoteJVMMaxMemMap    = -1;
 	public static long _remoteJVMMaxMemReduce = -1;
 	public static long _remoteMRSortMem = -1;
 	public static boolean _localJT      = false;
 	public static long _blocksize       = -1;

 	// Map from StatementBlock.ID to remoteJVMMaxMem (in bytes)
 	// Encodes MR job memory settings in the execution plan, if not found here, use the default MR setting in _remoteJVMMaxMem
 	public static HashMap<Long, Long> remoteJVMMaxMemPlan = new HashMap<Long, Long>();
 	public static HashSet<Long> probedSb = new HashSet<Long>();

 	public static List<Long> nodesMaxPhySorted = null;		// Original maximum physical memory per node in Byte, sorted
 	public static List<Double> nodesMaxBudgetSorted = null;	// Converted to maximum budget per node in Byte, sorted
 	public static int minimumMRContainerPhyMB = -1;			// Suggested minimum mappers physical memory
 	public static long mrAMPhy = -1;						// The default physical memory size of MR AM

 	public static long clusterTotalMem = -1;
 	public static int clusterTotalNodes = -1;
 	public static int clusterTotalCores = -1;
 	public static long minimalPhyAllocate = -1;
 	public static long maximumPhyAllocate = -1;

 	//client for resource utilization updates
 	private static YarnClient _client = null;

 	//static initialization, called for each JVM (on each node)
 	static
 	{
 		//analyze local node properties
 		analyzeLocalMachine();

 		//analyze remote Hadoop cluster properties
 		//analyzeYarnCluster(true); //note: due to overhead - analyze on-demand
 	}

 	public static List<Long> getNodesMaxPhySorted() {
 		if (nodesMaxPhySorted == null)
 			analyzeYarnCluster(true);
 		return nodesMaxPhySorted;
 	}

 	public static List<Double> getNodesMaxBudgetSorted() {
 		if (nodesMaxBudgetSorted == null)
 			analyzeYarnCluster(true);
 		return nodesMaxBudgetSorted;
 	}

 	public static long getMRARPhy() {
 		if (mrAMPhy == -1)
 			analyzeYarnCluster(true);
 		return mrAMPhy;
 	}

 	public static long getClusterTotalMem() {
 		if (clusterTotalMem == -1)
 			analyzeYarnCluster(true);
 		return clusterTotalMem;
 	}

 	public static long getMaxPhyAllocate() {
 		if (maximumPhyAllocate == -1)
 			analyzeYarnCluster(true);
 		return maximumPhyAllocate;
 	}

 	public static int getMinMRContarinerPhyMB() {
 		if (minimumMRContainerPhyMB == -1)
 			analyzeYarnCluster(true);
 		return minimumMRContainerPhyMB;
 	}
 	///////
 	//methods for obtaining parallelism properties

 	/**
 	 * Gets the number of logical processors of the current node,
 	 * including hyper-threading if enabled.
 	 *
 	 * @return
 	 */
 	public static int getLocalParallelism()
 	{
 		return _localPar;
 	}

 	/**
 	 * Gets the number of cluster nodes (number of tasktrackers). If multiple tasktracker
 	 * are started per node, each tasktracker is viewed as individual node.
 	 *
 	 * @return
 	 */
 	public static int getRemoteParallelNodes()
 	{
 		if( _remotePar == -1 )
 			analyzeYarnCluster(true);

 		return _remotePar;
 	}

 	/**
 	 * Gets the total number of available map slots.
 	 *
 	 * @return
 	 */
 	public static int getRemoteParallelMapTasks(long jobLookupId)
 	{
 		if (clusterTotalCores == -1)
 			analyzeYarnCluster(true);
 		int ret = getRemoteParallelTasksGivenMem(getRemoteMaxMemoryMap(jobLookupId));
 		//System.out.print("  jvm size " + OptimizerUtils.toMB(getRemoteMaxMemory(jobLookupId)) + " -> " + ret + " map tasks");
 		if (ret >= clusterTotalCores * CPU_HYPER_FACTOR)
 			ret = clusterTotalCores * CPU_HYPER_FACTOR;

 		//System.out.println(jobLookupId + " got " + ret + " jvm = " + OptimizerUtils.toMB(getRemoteMaxMemoryMap(jobLookupId)));
 		return ret;
 	}

 	/**
 	 * Gets the total number of available reduce slots.
 	 *
 	 * @return
 	 */
 	public static int getRemoteParallelReduceTasks(long jobLookupId)
 	{
 		if (clusterTotalCores == -1)
 			analyzeYarnCluster(true);
 		int ret = getRemoteParallelTasksGivenMem(getRemoteMaxMemoryReduce(jobLookupId));
 		if (ret >= clusterTotalCores * CPU_HYPER_FACTOR)
 			ret = clusterTotalCores * CPU_HYPER_FACTOR;
 		return ret;
 	}

 	public static long getYarnPhyAllocate(long requestPhy) {
 		if (minimalPhyAllocate == -1)
 			analyzeYarnCluster(true);
 		if (requestPhy > maximumPhyAllocate)
 			throw new RuntimeException("Requested " + OptimizerUtils.toMB(requestPhy) +
 					"MB, while the maximum yarn allocate is " + OptimizerUtils.toMB(maximumPhyAllocate) + "MB");

 		long ret = (long) Math.ceil((double)requestPhy / minimalPhyAllocate);
 		ret = ret * minimalPhyAllocate;
 		if (ret > maximumPhyAllocate)
 			ret = maximumPhyAllocate;
 		return ret;
 	}

 	/**
 	 * Gets the totals number of parallel tasks given its max memory size.
 	 *
 	 * @return
 	 */
 	public static int getRemoteParallelTasksGivenMem(long remoteTaskJvmMemory) {
 		long taskPhy = getYarnPhyAllocate(ResourceOptimizer.jvmToPhy(remoteTaskJvmMemory, false));
 		long cpPhy = getYarnPhyAllocate(ResourceOptimizer.jvmToPhy(getLocalMaxMemory(), false));
 		long mrAMPhy = getYarnPhyAllocate(getMRARPhy());

 		if (nodesMaxPhySorted == null)
 			analyzeYarnCluster(true);

 		if( nodesMaxPhySorted.isEmpty() )
 			return -1;
 		if (nodesMaxPhySorted.size() == 1) {
 			long tmp = nodesMaxPhySorted.get(0) - cpPhy - mrAMPhy;
 			if (tmp < 0)
 				return -1;
 			return (int)(tmp / taskPhy);
 		}
 		// At least have two nodes
 		long first = nodesMaxPhySorted.get(0) - cpPhy;
 		long second = nodesMaxPhySorted.get(1);

 		if (first >= second)
 			first -= mrAMPhy;
 		else
 			second -= mrAMPhy;
 		if (first < 0 || second < 0)
 			return -1;
 		long taskCount = first / taskPhy + second / taskPhy;
 		int counter = 0;
 		for (Long node : nodesMaxPhySorted) {
 			if (counter++ < 2)
 				continue;	// skip first two nodes
 			taskCount += node / taskPhy;
 		}

 		//System.out.println(OptimizerUtils.toMB(cpPhy) + " " + OptimizerUtils.toMB(mrAMPhy) + " " + OptimizerUtils.toMB(taskPhy) + " " + OptimizerUtils.toMB(nodesMaxPhySorted.get(1)));
 		return (int)taskCount;
 	}

 	public static boolean checkValidMemPlan(boolean hasMRJob) {
 		if (nodesMaxPhySorted == null)
 			analyzeYarnCluster(true);
 		if (!hasMRJob)
 			return nodesMaxPhySorted.get(0) >= getYarnPhyAllocate(ResourceOptimizer.jvmToPhy(getLocalMaxMemory(), false));
 		return getRemoteParallelTasksGivenMem(getMaximumRemoteMaxMemory()) > 0;
 	}

 	///////
 	//methods for obtaining memory properties

 	/**
 	 * Gets the maximum memory [in bytes] of the current JVM.
 	 *
 	 * @return
 	 */
 	public static long getLocalMaxMemory()
 	{
 		return _localJVMMaxMem;
 	}

 	/**
 	 *
 	 * @param localMem
 	 */
 	public static void setLocalMaxMemory( long localMem )
 	{
 		_localJVMMaxMem = localMem;
 	}


 	/**
 	 * Gets the maximum memory [in bytes] of all given hadoop task memory settings.
 	 *
 	 * @return
 	 */
 	public static long getMaximumRemoteMaxMemory()
 	{
 		if( _remoteJVMMaxMemMap == -1 )
 			analyzeYarnCluster(true);

 		long ret = (_remoteJVMMaxMemMap > _remoteJVMMaxMemReduce) ? _remoteJVMMaxMemMap : _remoteJVMMaxMemReduce;
 		for (Map.Entry<Long, Long> entry : remoteJVMMaxMemPlan.entrySet()) {
 			if (ret < entry.getValue())
 				ret = entry.getValue();
 		}
 		return ret;
 	}

 	/**
 	 * Gets the maximum memory [in bytes] of a hadoop map task JVM.
 	 *
 	 * @return
 	 */
 	public static long getRemoteMaxMemoryMap(long jobLookupId)
 	{
 		if( _remoteJVMMaxMemMap == -1 )
 			analyzeYarnCluster(true);

 		long ret = getSpecifiedRemoteMaxMemory(jobLookupId);
 		if (ret == -1)
 			ret = _remoteJVMMaxMemMap;
 		return ret;
 	}

 	/**
 	 * Gets the maximum memory [in bytes] of a hadoop reduce task JVM.
 	 *
 	 * @return
 	 */
 	public static long getRemoteMaxMemoryReduce(long jobLookupId)
 	{
 		if( _remoteJVMMaxMemReduce == -1 )
 			analyzeYarnCluster(true);

 		long ret = getSpecifiedRemoteMaxMemory(jobLookupId);
 		if (ret == -1)
 			ret = _remoteJVMMaxMemReduce;
 		return ret;
 	}

 	/**
 	 * Gets the maximum memory [in bytes] of a hadoop task JVM.
 	 *
 	 * @return
 	 */
 	public static long getSpecifiedRemoteMaxMemory(long jobLookupId)
 	{
 		probedSb.add(jobLookupId);

 		// Look up specified MR job setting
 		Long ret = remoteJVMMaxMemPlan.get(jobLookupId);
 		if (ret != null)
 			return ret;

 		// Look up specified default MR setting
 		ret = remoteJVMMaxMemPlan.get((long)-1);
 		if (ret != null)
 			return ret;

 		// No specified setting found
 		return -1;
 	}

 	public static void setRemoteMaxMemPlan(HashMap<Long, Double> budgetMRPlan) {
 		remoteJVMMaxMemPlan.clear();
 		for (Map.Entry<Long, Double> entry : budgetMRPlan.entrySet()) {
 			long mapJvm = ResourceOptimizer.budgetToJvm(entry.getValue());
 			remoteJVMMaxMemPlan.put(entry.getKey(), mapJvm);
 		}
 	}

 	public static void resetSBProbedSet() {
 		probedSb.clear();
 	}

 	public static HashSet<Long> getSBProbedSet() {
 		return probedSb;
 	}

 	public static void printProbedSet(String message) {
 		ArrayList<Long> probed = new ArrayList<Long> (probedSb);
 		Collections.sort(probed);
 		System.out.print(message);
 		for (Long id : probed)
 			System.out.print(id + ",");
 		System.out.println();
 	}

 	/**
 	 * Gets the maximum memory requirement [in bytes] of a given hadoop job.
 	 *
 	 * @param conf
 	 * @return
 	 */
 	/*public static long getRemoteMaxMemory( JobConf job )
 	{
 		return (1024*1024) * Math.max(
 				               job.getMemoryForMapTask(),
 				               job.getMemoryForReduceTask() );
 	}*/

 	/**
 	 * Gets the maximum sort buffer memory requirement [in bytes] of a hadoop task.
 	 *
 	 * @return
 	 */
 	public static long getRemoteMaxMemorySortBuffer( )
 	{
 		if( _remoteMRSortMem == -1 )
 			analyzeYarnCluster(true);

 		return _remoteMRSortMem;
 	}

 	///////
 	//methods for obtaining constraints or respective defaults

 	/**
 	 * Gets the maximum local parallelism constraint.
 	 *
 	 * @return
 	 */
 	public static int getCkMaxCP()
 	{
 		//default value (if not specified)
 		return getLocalParallelism();
 	}

 	/**
 	 * Gets the maximum remote parallelism constraint
 	 *
 	 * @return
 	 */
 	public static int getCkMaxMR(long jobLookupId)
 	{
 		//default value (if not specified)
 		return getRemoteParallelMapTasks(jobLookupId);
 	}

 	/**
 	 * Gets the maximum memory constraint [in bytes].
 	 *
 	 * @return
 	 */
 	public static long getCmMax(long jobLookupId)
 	{
 		//default value (if not specified)
 		return Math.min(getLocalMaxMemory(), getRemoteMaxMemoryMap(jobLookupId));
 	}

 	/**
 	 * Gets the HDFS blocksize of the used cluster in bytes.
 	 *
 	 * @return
 	 */
 	public static long getHDFSBlockSize()
 	{
 		if( _blocksize == -1 )
 			analyzeYarnCluster(true);

 		return _blocksize;
 	}

 	/**
 	 *
 	 * @param javaOpts
 	 * @return
 	 */
 	public static long extractMaxMemoryOpt(String javaOpts)
 	{
 		long ret = -1; //mem in bytes

 		try
 		{
 			StringTokenizer st = new StringTokenizer( javaOpts, " " );
 			while( st.hasMoreTokens() )
 			{
 				String arg = st.nextToken();
 				if( !arg.startsWith("-Xmx") ) //search for max mem
 					continue;

 				arg = arg.substring(4); //cut off "-Xmx"
 				//parse number and unit
 				if ( arg.endsWith("g") || arg.endsWith("G") )
 					ret = Long.parseLong(arg.substring(0,arg.length()-1)) * 1024 * 1024 * 1024;
 				else if ( arg.endsWith("m") || arg.endsWith("M") )
 					ret = Long.parseLong(arg.substring(0,arg.length()-1)) * 1024 * 1024;
 				else if( arg.endsWith("k") || arg.endsWith("K") )
 					ret = Long.parseLong(arg.substring(0,arg.length()-1)) * 1024;
 				else
 					ret = Long.parseLong(arg.substring(0,arg.length()-2));
 			}

 			if( ret < 0 ) // no argument found
 			{
 				ret = DEFAULT_JVM_SIZE;
 			}
 		}
 		catch(Exception ex)
 		{
 			//if anything breaks during parsing (e.g., because args not specified correctly)
 			ret = DEFAULT_JVM_SIZE;
 		}

 		return ret;
 	}

 	/**
 	 *
 	 * @param job
 	 * @param key
 	 * @param bytes
 	 */
 	public static void setMaxMemoryOpt(JobConf job, String key, long bytes)
 	{
 		String javaOptsOld = job.get( key );
 		String javaOptsNew = null;

 		//StringTokenizer st = new StringTokenizer( javaOptsOld, " " );
 		String[] tokens = javaOptsOld.split(" "); //account also for no ' '
 		StringBuilder sb = new StringBuilder();
 		for( String arg : tokens )
 		{
 			if( arg.startsWith("-Xmx") ) //search for max mem
 			{
 				sb.append("-Xmx");
 				sb.append( (bytes/(1024*1024)) );
 				sb.append("M");
 			}
 			else
 				sb.append(arg);

 			sb.append(" ");
 		}
 		javaOptsNew = sb.toString().trim();
 		job.set(key, javaOptsNew);
 	}

 	///////
 	//internal methods for analysis

 	/**
 	 * Analyzes properties of local machine and JVM.
 	 */
 	private static void analyzeLocalMachine()
 	{
 		_localPar       = Runtime.getRuntime().availableProcessors();
 		_localJVMMaxMem = Runtime.getRuntime().maxMemory();
 	}

 	public static void analyzeYarnCluster(boolean verbose) {
 		YarnConfiguration conf = new YarnConfiguration();
 		YarnClient yarnClient = YarnClient.createYarnClient();
 		yarnClient.init(conf);
 		yarnClient.start();
 		analyzeYarnCluster(yarnClient, conf, verbose);
 	}

 	public static long getMinAllocationBytes()
 	{
 		if( minimalPhyAllocate < 0 )
 			analyzeYarnCluster(false);
 		return minimalPhyAllocate;
 	}

 	public static long getMaxAllocationBytes()
 	{
 		if( maximumPhyAllocate < 0 )
 			analyzeYarnCluster(false);
 		return maximumPhyAllocate;
 	}

 	public static long getNumCores()
 	{
 		if( clusterTotalCores < 0 )
 			analyzeYarnCluster(false);
 		return clusterTotalCores;
 	}

 	public static long getNumNodes()
 	{
 		if( clusterTotalNodes < 0 )
 			analyzeYarnCluster(false);
 		return clusterTotalNodes;
 	}

 	public static YarnClusterConfig getClusterConfig()
 	{
 		YarnClusterConfig cc = new YarnClusterConfig();
 		cc.setMinAllocationMB( getMinAllocationBytes()/(1024*1024) );
 		cc.setMaxAllocationMB( getMaxAllocationBytes()/(1024*1024) );
 		cc.setNumNodes( getNumNodes() );
 		cc.setNumCores( getNumCores()*CPU_HYPER_FACTOR );

 		return cc;
 	}

 	/**
 	 *
 	 * @return
 	 * @throws YarnException
 	 * @throws IOException
 	 */
 	public static double getClusterUtilization()
 		throws IOException
 	{
 		double util = 0;

 		try
 		{
 			if( _client == null )
 				_client = createYarnClient();
 			List<NodeReport> nodesReport = _client.getNodeReports();

 			double maxMem = 0;
 			double currMem = 0;
 			long maxCores = 0;
 			long currCores = 0;
 			for (NodeReport node : nodesReport) {
 				Resource max = node.getCapability();
 				Resource used = node.getUsed();
 				maxMem += max.getMemory();
 				currMem += used.getMemory();
 				maxCores += max.getVirtualCores();
 				currCores += used.getVirtualCores();
 			}

 			util = Math.max(
 					  Math.min(1, currMem/maxMem),  //memory util
 					  Math.min(1, (double)currCores/maxCores) ); //vcore util
 		}
 		catch(Exception ex )
 		{
 			throw new IOException(ex);
 		}

 		return util;
 	}

 	/**
 	 * Analyzes properties of Yarn cluster and Hadoop configurations.
 	 */
 	public static void analyzeYarnCluster(YarnClient yarnClient, YarnConfiguration conf, boolean verbose) {
 		try {
 			List<NodeReport> nodesReport = yarnClient.getNodeReports();
 			if (verbose)
 				System.out.println("There are " + nodesReport.size() + " nodes in the cluster");
 			if( nodesReport.isEmpty() )
 				throw new YarnException("There are zero available nodes in the yarn cluster");

 			nodesMaxPhySorted = new ArrayList<Long> (nodesReport.size());
 			clusterTotalMem = 0;
 			clusterTotalCores = 0;
 			clusterTotalNodes = 0;
 			minimumMRContainerPhyMB = -1;
 			for (NodeReport node : nodesReport) {
 				Resource resource = node.getCapability();
 				Resource used = node.getUsed();
 				if (used == null)
 					used = Resource.newInstance(0, 0);
 				int mb = resource.getMemory();
 				int cores = resource.getVirtualCores();
 				if (mb <= 0)
 					throw new YarnException("A node has non-positive memory " + mb);

 				int myMinMRPhyMB = mb / cores / CPU_HYPER_FACTOR;
 				if (minimumMRContainerPhyMB < myMinMRPhyMB)
 					minimumMRContainerPhyMB = myMinMRPhyMB;	// minimumMRContainerPhyMB needs to be the largest among the mins

 				clusterTotalMem += (long)mb * 1024 * 1024;
 				nodesMaxPhySorted.add((long)mb * 1024 * 1024);
 				clusterTotalCores += cores;
 				clusterTotalNodes ++;
 				if (verbose)
 					System.out.println("\t" + node.getNodeId() + " has " + mb + " MB (" + used.getMemory() + " MB used) memory and " +
 						resource.getVirtualCores() + " (" + used.getVirtualCores() + " used) cores");

 			}
 			Collections.sort(nodesMaxPhySorted, Collections.reverseOrder());

 			nodesMaxBudgetSorted = new ArrayList<Double> (nodesMaxPhySorted.size());
 			for (int i = 0; i < nodesMaxPhySorted.size(); i++)
 				nodesMaxBudgetSorted.add(ResourceOptimizer.phyToBudget(nodesMaxPhySorted.get(i)));

 			_remotePar = nodesReport.size();
 			if (_remotePar == 0)
 				throw new YarnException("There are no available nodes in the yarn cluster");

 			// Now get the default cluster settings
 			_remoteMRSortMem = (1024*1024) * conf.getLong(MRConfigurationNames.MR_TASK_IO_SORT_MB,100); //100MB

 			//handle jvm max mem (map mem budget is relevant for map-side distcache and parfor)
 			//(for robustness we probe both: child and map configuration parameters)
 			String javaOpts1 = conf.get(MRConfigurationNames.MR_CHILD_JAVA_OPTS); //internally mapred/mapreduce synonym
 			String javaOpts2 = conf.get(MRConfigurationNames.MR_MAP_JAVA_OPTS, null); //internally mapred/mapreduce synonym
 			String javaOpts3 = conf.get(MRConfigurationNames.MR_REDUCE_JAVA_OPTS, null); //internally mapred/mapreduce synonym
 			if( javaOpts2 != null ) //specific value overrides generic
 				_remoteJVMMaxMemMap = extractMaxMemoryOpt(javaOpts2);
 			else
 				_remoteJVMMaxMemMap = extractMaxMemoryOpt(javaOpts1);
 			if( javaOpts3 != null ) //specific value overrides generic
 				_remoteJVMMaxMemReduce = extractMaxMemoryOpt(javaOpts3);
 			else
 				_remoteJVMMaxMemReduce = extractMaxMemoryOpt(javaOpts1);

 			//HDFS blocksize
 			String blocksize = conf.get(MRConfigurationNames.DFS_BLOCKSIZE, "134217728");
 			_blocksize = Long.parseLong(blocksize);

 			minimalPhyAllocate = (long) 1024 * 1024 * conf.getInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
 					YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB);
 			maximumPhyAllocate = (long) 1024 * 1024 * conf.getInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB,
 					YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB);
 			mrAMPhy = (long)conf.getInt(MRConfigurationNames.YARN_APP_MR_AM_RESOURCE_MB, 1536) * 1024 * 1024;

 		} catch (Exception e) {
 			throw new RuntimeException("Unable to analyze yarn cluster ", e);
 		}


 		/*
 		 * This is for AppMaster to query available resource in the cluster during heartbeat
 		 *
 		AMRMClient<ContainerRequest> rmClient = AMRMClient.createAMRMClient();
 		rmClient.init(conf);
 		rmClient.start();
 		AllocateResponse response = rmClient.allocate(0);
 		int nodeCount = response.getNumClusterNodes();
 		Resource resource = response.getAvailableResources();
 		List<NodeReport> nodeUpdate = response.getUpdatedNodes();

 		LOG.info("This is a " + nodeCount + " node cluster with totally " +
 				resource.getMemory() + " memory and " + resource.getVirtualCores() + " cores");
 		LOG.info(nodereport.size() + " updatedNode reports received");
 		for (NodeReport node : nodeUpdate) {
 			resource = node.getCapability();
 			LOG.info(node.getNodeId() + " updated with " + resource.getMemory() + " memory and " + resource.getVirtualCores() + " cores");
 		}*/
 	}

 	/**
 	 *
 	 * @return
 	 */
 	private static YarnClient createYarnClient()
 	{
 		YarnConfiguration conf = new YarnConfiguration();
 		YarnClient yarnClient = YarnClient.createYarnClient();
 		yarnClient.init(conf);
 		yarnClient.start();
 		return yarnClient;
 	}
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	package org.apache.sysml.yarn.ropt;

	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.StringTokenizer;

	import org.apache.hadoop.mapred.JobConf;
	import org.apache.hadoop.yarn.api.records.NodeReport;
	import org.apache.hadoop.yarn.api.records.Resource;
	import org.apache.hadoop.yarn.client.api.YarnClient;
	import org.apache.hadoop.yarn.conf.YarnConfiguration;
	import org.apache.hadoop.yarn.exceptions.YarnException;

	import org.apache.sysml.hops.OptimizerUtils;
	import org.apache.sysml.runtime.matrix.mapred.MRConfigurationNames;

	/**
	* Central place for analyzing and obtaining static infrastructure properties
	* such as memory and number of logical processors.
	*
	*
	*/
	public class YarnClusterAnalyzer
	{

	public static final long DEFAULT_JVM_SIZE = 512 * 1024 * 1024;
	public static final int CPU_HYPER_FACTOR = 1;

	//static local master node properties
	public static int _localPar = -1;
	public static long _localJVMMaxMem = -1;

	//default hadoop cluster properties
	public static int _remotePar = -1;
	//public static int _remoteParMap = -1;
	//public static int _remoteParReduce = -1;
	public static long _remoteJVMMaxMemMap = -1;
	public static long _remoteJVMMaxMemReduce = -1;
	public static long _remoteMRSortMem = -1;
	public static boolean _localJT = false;
	public static long _blocksize = -1;

	// Map from StatementBlock.ID to remoteJVMMaxMem (in bytes)
	// Encodes MR job memory settings in the execution plan, if not found here, use the default MR setting in _remoteJVMMaxMem
	public static HashMap<Long, Long> remoteJVMMaxMemPlan = new HashMap<Long, Long>();
	public static HashSet<Long> probedSb = new HashSet<Long>();

	public static List<Long> nodesMaxPhySorted = null; // Original maximum physical memory per node in Byte, sorted
	public static List<Double> nodesMaxBudgetSorted = null; // Converted to maximum budget per node in Byte, sorted
	public static int minimumMRContainerPhyMB = -1; // Suggested minimum mappers physical memory
	public static long mrAMPhy = -1; // The default physical memory size of MR AM

	public static long clusterTotalMem = -1;
	public static int clusterTotalNodes = -1;
	public static int clusterTotalCores = -1;
	public static long minimalPhyAllocate = -1;
	public static long maximumPhyAllocate = -1;

	//client for resource utilization updates
	private static YarnClient _client = null;

	//static initialization, called for each JVM (on each node)
	static
	{
	//analyze local node properties
	analyzeLocalMachine();

	//analyze remote Hadoop cluster properties
	//analyzeYarnCluster(true); //note: due to overhead - analyze on-demand
	}

	public static List<Long> getNodesMaxPhySorted() {
	if (nodesMaxPhySorted == null)
	analyzeYarnCluster(true);
	return nodesMaxPhySorted;
	}

	public static List<Double> getNodesMaxBudgetSorted() {
	if (nodesMaxBudgetSorted == null)
	analyzeYarnCluster(true);
	return nodesMaxBudgetSorted;
	}

	public static long getMRARPhy() {
	if (mrAMPhy == -1)
	analyzeYarnCluster(true);
	return mrAMPhy;
	}

	public static long getClusterTotalMem() {
	if (clusterTotalMem == -1)
	analyzeYarnCluster(true);
	return clusterTotalMem;
	}

	public static long getMaxPhyAllocate() {
	if (maximumPhyAllocate == -1)
	analyzeYarnCluster(true);
	return maximumPhyAllocate;
	}

	public static int getMinMRContarinerPhyMB() {
	if (minimumMRContainerPhyMB == -1)
	analyzeYarnCluster(true);
	return minimumMRContainerPhyMB;
	}
	///////
	//methods for obtaining parallelism properties

	/**
	* Gets the number of logical processors of the current node,
	* including hyper-threading if enabled.
	*
	* @return
	*/
	public static int getLocalParallelism()
	{
	return _localPar;
	}

	/**
	* Gets the number of cluster nodes (number of tasktrackers). If multiple tasktracker
	* are started per node, each tasktracker is viewed as individual node.
	*
	* @return
	*/
	public static int getRemoteParallelNodes()
	{
	if( _remotePar == -1 )
	analyzeYarnCluster(true);

	return _remotePar;
	}

	/**
	* Gets the total number of available map slots.
	*
	* @return
	*/
	public static int getRemoteParallelMapTasks(long jobLookupId)
	{
	if (clusterTotalCores == -1)
	analyzeYarnCluster(true);
	int ret = getRemoteParallelTasksGivenMem(getRemoteMaxMemoryMap(jobLookupId));
	//System.out.print(" jvm size " + OptimizerUtils.toMB(getRemoteMaxMemory(jobLookupId)) + " -> " + ret + " map tasks");
	if (ret >= clusterTotalCores * CPU_HYPER_FACTOR)
	ret = clusterTotalCores * CPU_HYPER_FACTOR;

	//System.out.println(jobLookupId + " got " + ret + " jvm = " + OptimizerUtils.toMB(getRemoteMaxMemoryMap(jobLookupId)));
	return ret;
	}

	/**
	* Gets the total number of available reduce slots.
	*
	* @return
	*/
	public static int getRemoteParallelReduceTasks(long jobLookupId)
	{
	if (clusterTotalCores == -1)
	analyzeYarnCluster(true);
	int ret = getRemoteParallelTasksGivenMem(getRemoteMaxMemoryReduce(jobLookupId));
	if (ret >= clusterTotalCores * CPU_HYPER_FACTOR)
	ret = clusterTotalCores * CPU_HYPER_FACTOR;
	return ret;
	}

	public static long getYarnPhyAllocate(long requestPhy) {
	if (minimalPhyAllocate == -1)
	analyzeYarnCluster(true);
	if (requestPhy > maximumPhyAllocate)
	throw new RuntimeException("Requested " + OptimizerUtils.toMB(requestPhy) +
	"MB, while the maximum yarn allocate is " + OptimizerUtils.toMB(maximumPhyAllocate) + "MB");

	long ret = (long) Math.ceil((double)requestPhy / minimalPhyAllocate);
	ret = ret * minimalPhyAllocate;
	if (ret > maximumPhyAllocate)
	ret = maximumPhyAllocate;
	return ret;
	}

	/**
	* Gets the totals number of parallel tasks given its max memory size.
	*
	* @return
	*/
	public static int getRemoteParallelTasksGivenMem(long remoteTaskJvmMemory) {
	long taskPhy = getYarnPhyAllocate(ResourceOptimizer.jvmToPhy(remoteTaskJvmMemory, false));
	long cpPhy = getYarnPhyAllocate(ResourceOptimizer.jvmToPhy(getLocalMaxMemory(), false));
	long mrAMPhy = getYarnPhyAllocate(getMRARPhy());

	if (nodesMaxPhySorted == null)
	analyzeYarnCluster(true);

	if( nodesMaxPhySorted.isEmpty() )
	return -1;
	if (nodesMaxPhySorted.size() == 1) {
	long tmp = nodesMaxPhySorted.get(0) - cpPhy - mrAMPhy;
	if (tmp < 0)
	return -1;
	return (int)(tmp / taskPhy);
	}
	// At least have two nodes
	long first = nodesMaxPhySorted.get(0) - cpPhy;
	long second = nodesMaxPhySorted.get(1);

	if (first >= second)
	first -= mrAMPhy;
	else
	second -= mrAMPhy;
	if (first < 0 \|\| second < 0)
	return -1;
	long taskCount = first / taskPhy + second / taskPhy;
	int counter = 0;
	for (Long node : nodesMaxPhySorted) {
	if (counter++ < 2)
	continue; // skip first two nodes
	taskCount += node / taskPhy;
	}

	//System.out.println(OptimizerUtils.toMB(cpPhy) + " " + OptimizerUtils.toMB(mrAMPhy) + " " + OptimizerUtils.toMB(taskPhy) + " " + OptimizerUtils.toMB(nodesMaxPhySorted.get(1)));
	return (int)taskCount;
	}

	public static boolean checkValidMemPlan(boolean hasMRJob) {
	if (nodesMaxPhySorted == null)
	analyzeYarnCluster(true);
	if (!hasMRJob)
	return nodesMaxPhySorted.get(0) >= getYarnPhyAllocate(ResourceOptimizer.jvmToPhy(getLocalMaxMemory(), false));
	return getRemoteParallelTasksGivenMem(getMaximumRemoteMaxMemory()) > 0;
	}

	///////
	//methods for obtaining memory properties

	/**
	* Gets the maximum memory [in bytes] of the current JVM.
	*
	* @return
	*/
	public static long getLocalMaxMemory()
	{
	return _localJVMMaxMem;
	}

	/**
	*
	* @param localMem
	*/
	public static void setLocalMaxMemory( long localMem )
	{
	_localJVMMaxMem = localMem;
	}


	/**
	* Gets the maximum memory [in bytes] of all given hadoop task memory settings.
	*
	* @return
	*/
	public static long getMaximumRemoteMaxMemory()
	{
	if( _remoteJVMMaxMemMap == -1 )
	analyzeYarnCluster(true);

	long ret = (_remoteJVMMaxMemMap > _remoteJVMMaxMemReduce) ? _remoteJVMMaxMemMap : _remoteJVMMaxMemReduce;
	for (Map.Entry<Long, Long> entry : remoteJVMMaxMemPlan.entrySet()) {
	if (ret < entry.getValue())
	ret = entry.getValue();
	}
	return ret;
	}

	/**
	* Gets the maximum memory [in bytes] of a hadoop map task JVM.
	*
	* @return
	*/
	public static long getRemoteMaxMemoryMap(long jobLookupId)
	{
	if( _remoteJVMMaxMemMap == -1 )
	analyzeYarnCluster(true);

	long ret = getSpecifiedRemoteMaxMemory(jobLookupId);
	if (ret == -1)
	ret = _remoteJVMMaxMemMap;
	return ret;
	}

	/**
	* Gets the maximum memory [in bytes] of a hadoop reduce task JVM.
	*
	* @return
	*/
	public static long getRemoteMaxMemoryReduce(long jobLookupId)
	{
	if( _remoteJVMMaxMemReduce == -1 )
	analyzeYarnCluster(true);

	long ret = getSpecifiedRemoteMaxMemory(jobLookupId);
	if (ret == -1)
	ret = _remoteJVMMaxMemReduce;
	return ret;
	}

	/**
	* Gets the maximum memory [in bytes] of a hadoop task JVM.
	*
	* @return
	*/
	public static long getSpecifiedRemoteMaxMemory(long jobLookupId)
	{
	probedSb.add(jobLookupId);

	// Look up specified MR job setting
	Long ret = remoteJVMMaxMemPlan.get(jobLookupId);
	if (ret != null)
	return ret;

	// Look up specified default MR setting
	ret = remoteJVMMaxMemPlan.get((long)-1);
	if (ret != null)
	return ret;

	// No specified setting found
	return -1;
	}

	public static void setRemoteMaxMemPlan(HashMap<Long, Double> budgetMRPlan) {
	remoteJVMMaxMemPlan.clear();
	for (Map.Entry<Long, Double> entry : budgetMRPlan.entrySet()) {
	long mapJvm = ResourceOptimizer.budgetToJvm(entry.getValue());
	remoteJVMMaxMemPlan.put(entry.getKey(), mapJvm);
	}
	}

	public static void resetSBProbedSet() {
	probedSb.clear();
	}

	public static HashSet<Long> getSBProbedSet() {
	return probedSb;
	}

	public static void printProbedSet(String message) {
	ArrayList<Long> probed = new ArrayList<Long> (probedSb);
	Collections.sort(probed);
	System.out.print(message);
	for (Long id : probed)
	System.out.print(id + ",");
	System.out.println();
	}

	/**
	* Gets the maximum memory requirement [in bytes] of a given hadoop job.
	*
	* @param conf
	* @return
	*/
	/*public static long getRemoteMaxMemory( JobConf job )
	{
	return (10241024) Math.max(
	job.getMemoryForMapTask(),
	job.getMemoryForReduceTask() );
	}*/

	/**
	* Gets the maximum sort buffer memory requirement [in bytes] of a hadoop task.
	*
	* @return
	*/
	public static long getRemoteMaxMemorySortBuffer( )
	{
	if( _remoteMRSortMem == -1 )
	analyzeYarnCluster(true);

	return _remoteMRSortMem;
	}

	///////
	//methods for obtaining constraints or respective defaults

	/**
	* Gets the maximum local parallelism constraint.
	*
	* @return
	*/
	public static int getCkMaxCP()
	{
	//default value (if not specified)
	return getLocalParallelism();
	}

	/**
	* Gets the maximum remote parallelism constraint
	*
	* @return
	*/
	public static int getCkMaxMR(long jobLookupId)
	{
	//default value (if not specified)
	return getRemoteParallelMapTasks(jobLookupId);
	}

	/**
	* Gets the maximum memory constraint [in bytes].
	*
	* @return
	*/
	public static long getCmMax(long jobLookupId)
	{
	//default value (if not specified)
	return Math.min(getLocalMaxMemory(), getRemoteMaxMemoryMap(jobLookupId));
	}

	/**
	* Gets the HDFS blocksize of the used cluster in bytes.
	*
	* @return
	*/
	public static long getHDFSBlockSize()
	{
	if( _blocksize == -1 )
	analyzeYarnCluster(true);

	return _blocksize;
	}

	/**
	*
	* @param javaOpts
	* @return
	*/
	public static long extractMaxMemoryOpt(String javaOpts)
	{
	long ret = -1; //mem in bytes

	try
	{
	StringTokenizer st = new StringTokenizer( javaOpts, " " );
	while( st.hasMoreTokens() )
	{
	String arg = st.nextToken();
	if( !arg.startsWith("-Xmx") ) //search for max mem
	continue;

	arg = arg.substring(4); //cut off "-Xmx"
	//parse number and unit
	if ( arg.endsWith("g") \|\| arg.endsWith("G") )
	ret = Long.parseLong(arg.substring(0,arg.length()-1)) * 1024 * 1024 * 1024;
	else if ( arg.endsWith("m") \|\| arg.endsWith("M") )
	ret = Long.parseLong(arg.substring(0,arg.length()-1)) * 1024 * 1024;
	else if( arg.endsWith("k") \|\| arg.endsWith("K") )
	ret = Long.parseLong(arg.substring(0,arg.length()-1)) * 1024;
	else
	ret = Long.parseLong(arg.substring(0,arg.length()-2));
	}

	if( ret < 0 ) // no argument found
	{
	ret = DEFAULT_JVM_SIZE;
	}
	}
	catch(Exception ex)
	{
	//if anything breaks during parsing (e.g., because args not specified correctly)
	ret = DEFAULT_JVM_SIZE;
	}

	return ret;
	}

	/**
	*
	* @param job
	* @param key
	* @param bytes
	*/
	public static void setMaxMemoryOpt(JobConf job, String key, long bytes)
	{
	String javaOptsOld = job.get( key );
	String javaOptsNew = null;

	//StringTokenizer st = new StringTokenizer( javaOptsOld, " " );
	String[] tokens = javaOptsOld.split(" "); //account also for no ' '
	StringBuilder sb = new StringBuilder();
	for( String arg : tokens )
	{
	if( arg.startsWith("-Xmx") ) //search for max mem
	{
	sb.append("-Xmx");
	sb.append( (bytes/(1024*1024)) );
	sb.append("M");
	}
	else
	sb.append(arg);

	sb.append(" ");
	}
	javaOptsNew = sb.toString().trim();
	job.set(key, javaOptsNew);
	}

	///////
	//internal methods for analysis

	/**
	* Analyzes properties of local machine and JVM.
	*/
	private static void analyzeLocalMachine()
	{
	_localPar = Runtime.getRuntime().availableProcessors();
	_localJVMMaxMem = Runtime.getRuntime().maxMemory();
	}

	public static void analyzeYarnCluster(boolean verbose) {
	YarnConfiguration conf = new YarnConfiguration();
	YarnClient yarnClient = YarnClient.createYarnClient();
	yarnClient.init(conf);
	yarnClient.start();
	analyzeYarnCluster(yarnClient, conf, verbose);
	}

	public static long getMinAllocationBytes()
	{
	if( minimalPhyAllocate < 0 )
	analyzeYarnCluster(false);
	return minimalPhyAllocate;
	}

	public static long getMaxAllocationBytes()
	{
	if( maximumPhyAllocate < 0 )
	analyzeYarnCluster(false);
	return maximumPhyAllocate;
	}

	public static long getNumCores()
	{
	if( clusterTotalCores < 0 )
	analyzeYarnCluster(false);
	return clusterTotalCores;
	}

	public static long getNumNodes()
	{
	if( clusterTotalNodes < 0 )
	analyzeYarnCluster(false);
	return clusterTotalNodes;
	}

	public static YarnClusterConfig getClusterConfig()
	{
	YarnClusterConfig cc = new YarnClusterConfig();
	cc.setMinAllocationMB( getMinAllocationBytes()/(1024*1024) );
	cc.setMaxAllocationMB( getMaxAllocationBytes()/(1024*1024) );
	cc.setNumNodes( getNumNodes() );
	cc.setNumCores( getNumCores()*CPU_HYPER_FACTOR );

	return cc;
	}

	/**
	*
	* @return
	* @throws YarnException
	* @throws IOException
	*/
	public static double getClusterUtilization()
	throws IOException
	{
	double util = 0;

	try
	{
	if( _client == null )
	_client = createYarnClient();
	List<NodeReport> nodesReport = _client.getNodeReports();

	double maxMem = 0;
	double currMem = 0;
	long maxCores = 0;
	long currCores = 0;
	for (NodeReport node : nodesReport) {
	Resource max = node.getCapability();
	Resource used = node.getUsed();
	maxMem += max.getMemory();
	currMem += used.getMemory();
	maxCores += max.getVirtualCores();
	currCores += used.getVirtualCores();
	}

	util = Math.max(
	Math.min(1, currMem/maxMem), //memory util
	Math.min(1, (double)currCores/maxCores) ); //vcore util
	}
	catch(Exception ex )
	{
	throw new IOException(ex);
	}

	return util;
	}

	/**
	* Analyzes properties of Yarn cluster and Hadoop configurations.
	*/
	public static void analyzeYarnCluster(YarnClient yarnClient, YarnConfiguration conf, boolean verbose) {
	try {
	List<NodeReport> nodesReport = yarnClient.getNodeReports();
	if (verbose)
	System.out.println("There are " + nodesReport.size() + " nodes in the cluster");
	if( nodesReport.isEmpty() )
	throw new YarnException("There are zero available nodes in the yarn cluster");

	nodesMaxPhySorted = new ArrayList<Long> (nodesReport.size());
	clusterTotalMem = 0;
	clusterTotalCores = 0;
	clusterTotalNodes = 0;
	minimumMRContainerPhyMB = -1;
	for (NodeReport node : nodesReport) {
	Resource resource = node.getCapability();
	Resource used = node.getUsed();
	if (used == null)
	used = Resource.newInstance(0, 0);
	int mb = resource.getMemory();
	int cores = resource.getVirtualCores();
	if (mb <= 0)
	throw new YarnException("A node has non-positive memory " + mb);

	int myMinMRPhyMB = mb / cores / CPU_HYPER_FACTOR;
	if (minimumMRContainerPhyMB < myMinMRPhyMB)
	minimumMRContainerPhyMB = myMinMRPhyMB; // minimumMRContainerPhyMB needs to be the largest among the mins

	clusterTotalMem += (long)mb * 1024 * 1024;
	nodesMaxPhySorted.add((long)mb * 1024 * 1024);
	clusterTotalCores += cores;
	clusterTotalNodes ++;
	if (verbose)
	System.out.println("\t" + node.getNodeId() + " has " + mb + " MB (" + used.getMemory() + " MB used) memory and " +
	resource.getVirtualCores() + " (" + used.getVirtualCores() + " used) cores");

	}
	Collections.sort(nodesMaxPhySorted, Collections.reverseOrder());

	nodesMaxBudgetSorted = new ArrayList<Double> (nodesMaxPhySorted.size());
	for (int i = 0; i < nodesMaxPhySorted.size(); i++)
	nodesMaxBudgetSorted.add(ResourceOptimizer.phyToBudget(nodesMaxPhySorted.get(i)));

	_remotePar = nodesReport.size();
	if (_remotePar == 0)
	throw new YarnException("There are no available nodes in the yarn cluster");

	// Now get the default cluster settings
	_remoteMRSortMem = (10241024) conf.getLong(MRConfigurationNames.MR_TASK_IO_SORT_MB,100); //100MB

	//handle jvm max mem (map mem budget is relevant for map-side distcache and parfor)
	//(for robustness we probe both: child and map configuration parameters)
	String javaOpts1 = conf.get(MRConfigurationNames.MR_CHILD_JAVA_OPTS); //internally mapred/mapreduce synonym
	String javaOpts2 = conf.get(MRConfigurationNames.MR_MAP_JAVA_OPTS, null); //internally mapred/mapreduce synonym
	String javaOpts3 = conf.get(MRConfigurationNames.MR_REDUCE_JAVA_OPTS, null); //internally mapred/mapreduce synonym
	if( javaOpts2 != null ) //specific value overrides generic
	_remoteJVMMaxMemMap = extractMaxMemoryOpt(javaOpts2);
	else
	_remoteJVMMaxMemMap = extractMaxMemoryOpt(javaOpts1);
	if( javaOpts3 != null ) //specific value overrides generic
	_remoteJVMMaxMemReduce = extractMaxMemoryOpt(javaOpts3);
	else
	_remoteJVMMaxMemReduce = extractMaxMemoryOpt(javaOpts1);

	//HDFS blocksize
	String blocksize = conf.get(MRConfigurationNames.DFS_BLOCKSIZE, "134217728");
	_blocksize = Long.parseLong(blocksize);

	minimalPhyAllocate = (long) 1024 * 1024 * conf.getInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
	YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB);
	maximumPhyAllocate = (long) 1024 * 1024 * conf.getInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB,
	YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB);
	mrAMPhy = (long)conf.getInt(MRConfigurationNames.YARN_APP_MR_AM_RESOURCE_MB, 1536) * 1024 * 1024;

	} catch (Exception e) {
	throw new RuntimeException("Unable to analyze yarn cluster ", e);
	}



	/*
	* This is for AppMaster to query available resource in the cluster during heartbeat
	*
	AMRMClient<ContainerRequest> rmClient = AMRMClient.createAMRMClient();
	rmClient.init(conf);
	rmClient.start();
	AllocateResponse response = rmClient.allocate(0);
	int nodeCount = response.getNumClusterNodes();
	Resource resource = response.getAvailableResources();
	List<NodeReport> nodeUpdate = response.getUpdatedNodes();

	LOG.info("This is a " + nodeCount + " node cluster with totally " +
	resource.getMemory() + " memory and " + resource.getVirtualCores() + " cores");
	LOG.info(nodereport.size() + " updatedNode reports received");
	for (NodeReport node : nodeUpdate) {
	resource = node.getCapability();
	LOG.info(node.getNodeId() + " updated with " + resource.getMemory() + " memory and " + resource.getVirtualCores() + " cores");
	}*/
	}

	/**
	*
	* @return
	*/
	private static YarnClient createYarnClient()
	{
	YarnConfiguration conf = new YarnConfiguration();
	YarnClient yarnClient = YarnClient.createYarnClient();
	yarnClient.init(conf);
	yarnClient.start();
	return yarnClient;
	}
	}