MAPREDUCE-5398. MR changes for YARN-513 (Jian He via bikas)
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1503934 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
33b5a81509
commit
bf876f70fb
@ -344,6 +344,8 @@ Release 2.1.0-beta - 2013-07-02
|
|||||||
MAPREDUCE-5333. Add test that verifies MRAM works correctly when sending
|
MAPREDUCE-5333. Add test that verifies MRAM works correctly when sending
|
||||||
requests with non-normalized capabilities. (ywskycn via tucu)
|
requests with non-normalized capabilities. (ywskycn via tucu)
|
||||||
|
|
||||||
|
MAPREDUCE-5398. MR changes for YARN-513 (Jian He via bikas)
|
||||||
|
|
||||||
OPTIMIZATIONS
|
OPTIMIZATIONS
|
||||||
|
|
||||||
MAPREDUCE-4974. Optimising the LineRecordReader initialize() method
|
MAPREDUCE-4974. Optimising the LineRecordReader initialize() method
|
||||||
|
@ -21,7 +21,6 @@
|
|||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.net.InetSocketAddress;
|
import java.net.InetSocketAddress;
|
||||||
import java.nio.ByteBuffer;
|
import java.nio.ByteBuffer;
|
||||||
import java.security.PrivilegedAction;
|
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||||
import java.util.concurrent.atomic.AtomicBoolean;
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
@ -50,12 +49,11 @@
|
|||||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||||
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
|
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
|
||||||
import org.apache.hadoop.yarn.api.records.Resource;
|
import org.apache.hadoop.yarn.api.records.Resource;
|
||||||
import org.apache.hadoop.yarn.conf.YarnConfiguration;
|
import org.apache.hadoop.yarn.client.ClientRMProxy;
|
||||||
import org.apache.hadoop.yarn.event.EventHandler;
|
import org.apache.hadoop.yarn.event.EventHandler;
|
||||||
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
|
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
|
||||||
import org.apache.hadoop.yarn.factories.RecordFactory;
|
import org.apache.hadoop.yarn.factories.RecordFactory;
|
||||||
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
|
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
|
||||||
import org.apache.hadoop.yarn.ipc.YarnRPC;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Registers/unregisters to RM and sends heartbeats to RM.
|
* Registers/unregisters to RM and sends heartbeats to RM.
|
||||||
@ -265,27 +263,12 @@ public void run() {
|
|||||||
|
|
||||||
protected ApplicationMasterProtocol createSchedulerProxy() {
|
protected ApplicationMasterProtocol createSchedulerProxy() {
|
||||||
final Configuration conf = getConfig();
|
final Configuration conf = getConfig();
|
||||||
final YarnRPC rpc = YarnRPC.create(conf);
|
|
||||||
final InetSocketAddress serviceAddr = conf.getSocketAddr(
|
|
||||||
YarnConfiguration.RM_SCHEDULER_ADDRESS,
|
|
||||||
YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS,
|
|
||||||
YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT);
|
|
||||||
|
|
||||||
UserGroupInformation currentUser;
|
|
||||||
try {
|
try {
|
||||||
currentUser = UserGroupInformation.getCurrentUser();
|
return ClientRMProxy.createRMProxy(conf, ApplicationMasterProtocol.class);
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
throw new YarnRuntimeException(e);
|
throw new YarnRuntimeException(e);
|
||||||
}
|
}
|
||||||
|
|
||||||
// CurrentUser should already have AMToken loaded.
|
|
||||||
return currentUser.doAs(new PrivilegedAction<ApplicationMasterProtocol>() {
|
|
||||||
@Override
|
|
||||||
public ApplicationMasterProtocol run() {
|
|
||||||
return (ApplicationMasterProtocol) rpc.getProxy(ApplicationMasterProtocol.class,
|
|
||||||
serviceAddr, conf);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
}
|
||||||
|
|
||||||
protected abstract void heartbeat() throws Exception;
|
protected abstract void heartbeat() throws Exception;
|
||||||
|
@ -75,36 +75,18 @@ public class ResourceMgrDelegate extends YarnClient {
|
|||||||
* @param conf the configuration object.
|
* @param conf the configuration object.
|
||||||
*/
|
*/
|
||||||
public ResourceMgrDelegate(YarnConfiguration conf) {
|
public ResourceMgrDelegate(YarnConfiguration conf) {
|
||||||
this(conf, null);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Delegate responsible for communicating with the Resource Manager's
|
|
||||||
* {@link ApplicationClientProtocol}.
|
|
||||||
* @param conf the configuration object.
|
|
||||||
* @param rmAddress the address of the Resource Manager
|
|
||||||
*/
|
|
||||||
public ResourceMgrDelegate(YarnConfiguration conf,
|
|
||||||
InetSocketAddress rmAddress) {
|
|
||||||
super(ResourceMgrDelegate.class.getName());
|
super(ResourceMgrDelegate.class.getName());
|
||||||
this.conf = conf;
|
this.conf = conf;
|
||||||
this.rmAddress = rmAddress;
|
this.client = YarnClient.createYarnClient();
|
||||||
if (rmAddress == null) {
|
|
||||||
client = YarnClient.createYarnClient();
|
|
||||||
} else {
|
|
||||||
client = YarnClient.createYarnClient(rmAddress);
|
|
||||||
}
|
|
||||||
init(conf);
|
init(conf);
|
||||||
start();
|
start();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected void serviceInit(Configuration conf) throws Exception {
|
protected void serviceInit(Configuration conf) throws Exception {
|
||||||
if (rmAddress == null) {
|
this.rmAddress = conf.getSocketAddr(YarnConfiguration.RM_ADDRESS,
|
||||||
this.rmAddress = conf.getSocketAddr(YarnConfiguration.RM_ADDRESS,
|
|
||||||
YarnConfiguration.DEFAULT_RM_ADDRESS,
|
YarnConfiguration.DEFAULT_RM_ADDRESS,
|
||||||
YarnConfiguration.DEFAULT_RM_PORT);
|
YarnConfiguration.DEFAULT_RM_PORT);
|
||||||
}
|
|
||||||
client.init(conf);
|
client.init(conf);
|
||||||
super.serviceInit(conf);
|
super.serviceInit(conf);
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user