Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

MAPREDUCE-6697. Concurrent task limits should only be applied when necessary. Contributed by Nathan Roberts. 

(cherry picked from commit a5c0476a99)
This commit is contained in:
Akira Ajisaka 2017-06-28 10:50:09 +09:00 committed by Konstantin V Shvachko
parent 187bb77ddb
commit 7d24d4751e
3 changed files with 76 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-mapreduce-project/CHANGES.txt
View File

@ -41,6 +41,9 @@ Release 2.7.4 - UNRELEASED
MAPREDUCE-6433. launchTime may be negative.
(zxu) backported by Chris Douglas.
MAPREDUCE-6697. Concurrent task limits should only be applied when
necessary. (Nathan Roberts via shv).
Release 2.7.3 - 2016-08-25
INCOMPATIBLE CHANGES

6
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java
View File

@ -818,7 +818,8 @@ public class RMContainerAllocator extends RMContainerRequestor
private void applyConcurrentTaskLimits() {
int numScheduledMaps = scheduledRequests.maps.size();
if (maxRunningMaps > 0 && numScheduledMaps > 0) {
if (maxRunningMaps > 0 && numScheduledMaps > 0 &&
getJob().getTotalMaps() > maxRunningMaps) {
int maxRequestedMaps = Math.max(0,
maxRunningMaps - assignedRequests.maps.size());
int numScheduledFailMaps = scheduledRequests.earlierFailedMaps.size();
@ -833,7 +834,8 @@ public class RMContainerAllocator extends RMContainerRequestor
}
int numScheduledReduces = scheduledRequests.reduces.size();
if (maxRunningReduces > 0 && numScheduledReduces > 0) {
if (maxRunningReduces > 0 && numScheduledReduces > 0 &&
getJob().getTotalReduces() > maxRunningReduces) {
int maxRequestedReduces = Math.max(0,
maxRunningReduces - assignedRequests.reduces.size());
int reduceRequestLimit = Math.min(maxRequestedReduces,

73
hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java
View File

@ -2617,15 +2617,78 @@ public class TestRMContainerAllocator {
new Text(rmAddr), ugiToken.getService());
}
@Test
public void testConcurrentTaskLimitsDisabledIfSmaller() throws Exception {
final int MAP_COUNT = 1;
final int REDUCE_COUNT = 1;
final int MAP_LIMIT = 1;
final int REDUCE_LIMIT = 1;
Configuration conf = new Configuration();
conf.setInt(MRJobConfig.JOB_RUNNING_MAP_LIMIT, MAP_LIMIT);
conf.setInt(MRJobConfig.JOB_RUNNING_REDUCE_LIMIT, REDUCE_LIMIT);
conf.setFloat(MRJobConfig.COMPLETED_MAPS_FOR_REDUCE_SLOWSTART, 0.0f);
ApplicationId appId = ApplicationId.newInstance(1, 1);
ApplicationAttemptId appAttemptId =
ApplicationAttemptId.newInstance(appId, 1);
JobId jobId = MRBuilderUtils.newJobId(appAttemptId.getApplicationId(), 0);
Job mockJob = mock(Job.class);
when(mockJob.getReport()).thenReturn(
MRBuilderUtils.newJobReport(jobId, "job", "user", JobState.RUNNING, 0,
0, 0, 0, 0, 0, 0, "jobfile", null, false, ""));
when(mockJob.getTotalMaps()).thenReturn(MAP_COUNT);
when(mockJob.getTotalReduces()).thenReturn(REDUCE_COUNT);
final MockScheduler mockScheduler = new MockScheduler(appAttemptId);
MyContainerAllocator allocator =
new MyContainerAllocator(null, conf, appAttemptId, mockJob,
new SystemClock()) {
@Override
protected void register() {
}
@Override
protected ApplicationMasterProtocol createSchedulerProxy() {
return mockScheduler;
}
@Override
protected void setRequestLimit(Priority priority,
Resource capability, int limit) {
Assert.fail("setRequestLimit() should not be invoked");
}
};
// create some map requests
ContainerRequestEvent[] reqMapEvents = new ContainerRequestEvent[MAP_COUNT];
for (int i = 0; i < reqMapEvents.length; ++i) {
reqMapEvents[i] = createReq(jobId, i, 1024, new String[] { "h" + i });
}
allocator.sendRequests(Arrays.asList(reqMapEvents));
// create some reduce requests
ContainerRequestEvent[] reqReduceEvents =
new ContainerRequestEvent[REDUCE_COUNT];
for (int i = 0; i < reqReduceEvents.length; ++i) {
reqReduceEvents[i] =
createReq(jobId, i, 1024, new String[] {}, false, true);
}
allocator.sendRequests(Arrays.asList(reqReduceEvents));
allocator.schedule();
allocator.schedule();
allocator.schedule();
allocator.close();
}
@Test
public void testConcurrentTaskLimits() throws Exception {
final int MAP_COUNT = 5;
final int REDUCE_COUNT = 2;
final int MAP_LIMIT = 3;
final int REDUCE_LIMIT = 1;
LOG.info("Running testConcurrentTaskLimits");
Configuration conf = new Configuration();
conf.setInt(MRJobConfig.JOB_RUNNING_MAP_LIMIT, MAP_LIMIT);
conf.setInt(MRJobConfig.JOB_RUNNING_REDUCE_LIMIT, REDUCE_LIMIT);
conf.setFloat(MRJobConfig.COMPLETED_MAPS_FOR_REDUCE_SLOWSTART, 1.0f);
conf.setFloat(MRJobConfig.COMPLETED_MAPS_FOR_REDUCE_SLOWSTART, 0.0f);
ApplicationId appId = ApplicationId.newInstance(1, 1);
ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(
appId, 1);
@ -2634,6 +2697,9 @@ public class TestRMContainerAllocator {
when(mockJob.getReport()).thenReturn(
MRBuilderUtils.newJobReport(jobId, "job", "user", JobState.RUNNING, 0,
0, 0, 0, 0, 0, 0, "jobfile", null, false, ""));
when(mockJob.getTotalMaps()).thenReturn(MAP_COUNT);
when(mockJob.getTotalReduces()).thenReturn(REDUCE_COUNT);
final MockScheduler mockScheduler = new MockScheduler(appAttemptId);
MyContainerAllocator allocator = new MyContainerAllocator(null, conf,
appAttemptId, mockJob, new SystemClock()) {
@ -2648,14 +2714,13 @@ public class TestRMContainerAllocator {
};
// create some map requests
ContainerRequestEvent[] reqMapEvents = new ContainerRequestEvent[5];
ContainerRequestEvent[] reqMapEvents = new ContainerRequestEvent[MAP_COUNT];
for (int i = 0; i < reqMapEvents.length; ++i) {
reqMapEvents[i] = createReq(jobId, i, 1024, new String[] { "h" + i });
}
allocator.sendRequests(Arrays.asList(reqMapEvents));
// create some reduce requests
ContainerRequestEvent[] reqReduceEvents = new ContainerRequestEvent[2];
ContainerRequestEvent[] reqReduceEvents = new ContainerRequestEvent[REDUCE_COUNT];
for (int i = 0; i < reqReduceEvents.length; ++i) {
reqReduceEvents[i] = createReq(jobId, i, 1024, new String[] {},
false, true);