I have a patch that tries to use file links instead of making a copy of the
data that is already available locally. I tested it on the a single machine
cluster configuration running 48 mappers and reducers. I unfortunately do
not have access to a cluster even a small one. Can some on review and test
run my patch ?
I created the patch using Eclipse against 1.0.3. My knowledge in Java in
limited and the code is not well written in some classes. So please let me
know if I need to make changes to the code along with a short explanation
of the change. I will happily do so.
Thanks,
Suresh.
### Eclipse Workspace Patch 1.0
#P hadoop-1.0.4
Index: src/mapred/org/apache/hadoop/mapred/TaskCompletionEvent.java
===================================================================
--- src/mapred/org/apache/hadoop/mapred/TaskCompletionEvent.java
(revision 1401127)
+++ src/mapred/org/apache/hadoop/mapred/TaskCompletionEvent.java
(working copy)
@@ -30,6 +30,11 @@
* job tracker.
*/
public class TaskCompletionEvent implements Writable{
+
+ public String getTaskTrackerName() {
+ return taskTrackerName;
+ }
+
static public enum Status {FAILED, KILLED, SUCCEEDED, OBSOLETE, TIPFAILED};
private int eventId;
@@ -39,6 +44,12 @@
Status status;
boolean isMap = false;
private int idWithinJob;
+ private boolean isMapOutputLocal = false;
+ private String mapOutputLocation = null;
+ private long mapOutputOffSet = 0;
+ private long mapOutputReadLength = 0;
+ private long mapOutputRawLength = 0;
+ private String taskTrackerName = null;
public static final TaskCompletionEvent[] EMPTY_ARRAY =
new TaskCompletionEvent[0];
/**
@@ -63,7 +74,7 @@
int idWithinJob,
boolean isMap,
Status status,
- String taskTrackerHttp){
+ String taskTrackerHttp, String taskTrackerName){
this.taskId = taskId;
this.idWithinJob = idWithinJob;
@@ -71,7 +82,57 @@
this.eventId = eventId;
this.status =status;
this.taskTrackerHttp = taskTrackerHttp;
+ this.taskTrackerName = taskTrackerName;
+ }
+
+ public TaskCompletionEvent(int eventId,
+ TaskAttemptID taskId,
+ int idWithinJob,
+ boolean isMap,
+ Status status,
+ String taskTrackerHttp,
+ boolean sameTaskTrackerForMap, String mapOutputLocation, long
mapOutputOffSet, long mapOutputReadLength, long mapOutputUnCompressedLength,
String taskTrackerName){
+
+ this.taskId = taskId;
+ this.idWithinJob = idWithinJob;
+ this.isMap = isMap;
+ this.eventId = eventId;
+ this.status =status;
+ this.taskTrackerHttp = taskTrackerHttp;
+ this.isMapOutputLocal = sameTaskTrackerForMap;
+ this.mapOutputLocation = mapOutputLocation;
+ this.mapOutputOffSet = mapOutputOffSet;
+ this.mapOutputReadLength = mapOutputReadLength;
+ this.taskTrackerName = taskTrackerName;
+ this.mapOutputRawLength = mapOutputUnCompressedLength;
+ }
+
+ /**
+ * Returns true or false depending of if the map event occured on the same
task tracker as the reducer requesting the event
+ * @return boolean
+ */
+ public boolean isSameTaskTrackerForMap() {
+ return isMapOutputLocal;
+ }
+
+ public String getMapOutputLocation() {
+ return mapOutputLocation;
+ }
+
+ public long getMapOutputOffSet()
+ {
+ return mapOutputOffSet;
+ }
+
+ public long getMapOutputReadLength()
+ {
+ return mapOutputReadLength;
+ }
+
+ public long getMapOutputRawLength() {
+ return mapOutputRawLength;
}
+
/**
* Returns event Id.
* @return event id
@@ -128,6 +189,32 @@
}
/**
+ * Set true or false depending of if the map event occured on the same task
tracker as the reducer requesting the event
+ * @param sameTaskTrackerForMap boolean
+ */
+ public void setMapOutputLocal(boolean isMapOutputLocal) {
+ this.isMapOutputLocal = isMapOutputLocal;
+ }
+
+ public void setMapOutputLocation(String mapOutputLocation) {
+ this.mapOutputLocation = mapOutputLocation;
+ }
+
+ public void setMapOutputOffSet(long mapOutputOffSet)
+ {
+ this.mapOutputOffSet = mapOutputOffSet;
+ }
+
+ public void setMapOutputReadLength(long mapOutputReadLength)
+ {
+ this.mapOutputReadLength = mapOutputReadLength;
+ }
+
+ public void setMapOutputRawLength(long mapOutputUnCompressedLength) {
+ this.mapOutputRawLength = mapOutputUnCompressedLength;
+ }
+
+ /**
* set event Id. should be assigned incrementally starting from 0.
* @param eventId
*/
@@ -192,7 +279,12 @@
&& this.status.equals(event.getTaskStatus())
&& this.taskId.equals(event.getTaskAttemptId())
&& this.taskRunTime == event.getTaskRunTime()
- && this.taskTrackerHttp.equals(event.getTaskTrackerHttp());
+ && this.taskTrackerHttp.equals(event.getTaskTrackerHttp())
+ && this.isMapOutputLocal == event.isMapOutputLocal
+ && this.mapOutputLocation.equals(event.mapOutputLocation)
+ && this.mapOutputOffSet == event.mapOutputOffSet
+ && this.mapOutputReadLength == event.mapOutputReadLength
+ && this.mapOutputRawLength == event.mapOutputRawLength;
}
return false;
}
@@ -220,6 +312,12 @@
WritableUtils.writeString(out, taskTrackerHttp);
WritableUtils.writeVInt(out, taskRunTime);
WritableUtils.writeVInt(out, eventId);
+ out.writeBoolean(isMapOutputLocal);
+ WritableUtils.writeString(out, mapOutputLocation);
+ WritableUtils.writeVLong(out, mapOutputOffSet);
+ WritableUtils.writeVLong(out, mapOutputReadLength);
+ WritableUtils.writeVLong(out,mapOutputRawLength);
+ WritableUtils.writeString(out, taskTrackerName);
}
public void readFields(DataInput in) throws IOException {
@@ -230,5 +328,11 @@
taskTrackerHttp = WritableUtils.readString(in);
taskRunTime = WritableUtils.readVInt(in);
eventId = WritableUtils.readVInt(in);
+ isMapOutputLocal = in.readBoolean();
+ mapOutputLocation = WritableUtils.readString(in);
+ mapOutputOffSet = WritableUtils.readVLong(in);
+ mapOutputReadLength = WritableUtils.readVLong(in);
+ mapOutputRawLength = WritableUtils.readVLong(in);
+ taskTrackerName = WritableUtils.readString(in);
}
}
Index: src/mapred/org/apache/hadoop/mapred/JobInProgress.java
===================================================================
--- src/mapred/org/apache/hadoop/mapred/JobInProgress.java (revision
1401127)
+++ src/mapred/org/apache/hadoop/mapred/JobInProgress.java (working copy)
@@ -1121,7 +1121,8 @@
!tip.isJobCleanupTask() &&
!tip.isJobSetupTask(),
TaskCompletionEvent.Status.SUCCEEDED,
- httpTaskLogLocation
+ httpTaskLogLocation ,
+ taskTracker.getTrackerName()
);
taskEvent.setTaskRunTime((int)(status.getFinishTime()
- status.getStartTime()));
@@ -1177,7 +1178,8 @@
!tip.isJobCleanupTask() &&
!tip.isJobSetupTask(),
taskCompletionStatus,
- httpTaskLogLocation
+ httpTaskLogLocation,
+ taskTracker.getTrackerName()
);
}
Index: src/mapred/org/apache/hadoop/mapred/ReduceTask.java
===================================================================
--- src/mapred/org/apache/hadoop/mapred/ReduceTask.java (revision 1401127)
+++ src/mapred/org/apache/hadoop/mapred/ReduceTask.java (working copy)
@@ -21,6 +21,7 @@
import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
+import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -55,6 +56,7 @@
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileSystem.Statistics;
+import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DataInputBuffer;
@@ -140,7 +142,7 @@
return -1;
else if (a.getLen() == b.getLen())
if (a.getPath().toString().equals(b.getPath().toString()))
- return 0;
+ return (a.getOffset() == b.getOffset() ? 0 : (a.getOffset() <
b.getOffset()) ? 1 : -1);
else
return -1;
else
@@ -361,7 +363,6 @@
reporter.startCommunicationThread();
boolean useNewApi = job.getUseNewReducer();
initialize(job, getJobID(), reporter, useNewApi);
-
// check if it is a cleanupJobTask
if (jobCleanup) {
runJobCleanupTask(umbilical, reporter);
@@ -378,9 +379,11 @@
// Initialize the codec
codec = initCodec();
-
- boolean isLocal = "local".equals(job.get("mapred.job.tracker", "local"));
+
+ String trackerloc = job.get("mapred.job.tracker", "local");
+ boolean isLocal = "local".equals(trackerloc);
if (!isLocal) {
+ LOG.info(" The task is not a single threaded map-reduce job ");
reduceCopier = new ReduceCopier(umbilical, job, reporter);
if (!reduceCopier.fetchOutputs()) {
if(reduceCopier.mergeThrowable instanceof FSError) {
@@ -391,6 +394,7 @@
}
}
copyPhase.complete(); // copy is already complete
+ LOG.info("Copy phase is complete");
setPhase(TaskStatus.Phase.SORT);
statusUpdate(umbilical);
@@ -407,6 +411,7 @@
mapOutputFilesOnDisk.clear();
sortPhase.complete(); // sort is complete
+ LOG.info("Sort phase is complete");
setPhase(TaskStatus.Phase.REDUCE);
statusUpdate(umbilical);
Class keyClass = job.getMapOutputKeyClass();
@@ -998,10 +1003,14 @@
* Abstraction to track a map-output.
*/
private class MapOutputLocation {
+
TaskAttemptID taskAttemptId;
TaskID taskId;
String ttHost;
URL taskOutput;
+ long offSet;
+ long length;
+ long rawLength;
public MapOutputLocation(TaskAttemptID taskAttemptId,
String ttHost, URL taskOutput) {
@@ -1009,6 +1018,19 @@
this.taskId = this.taskAttemptId.getTaskID();
this.ttHost = ttHost;
this.taskOutput = taskOutput;
+ this.offSet = 0;
+ this.length = 0;
+ }
+
+ public MapOutputLocation(TaskAttemptID taskAttemptId,
+ String ttHost, URL taskOutput, long offSet, long length, long
rawLength) {
+ this.taskAttemptId = taskAttemptId;
+ this.taskId = this.taskAttemptId.getTaskID();
+ this.ttHost = ttHost;
+ this.taskOutput = taskOutput;
+ this.offSet = offSet;
+ this.length = length;
+ this.rawLength = rawLength;
}
public TaskAttemptID getTaskAttemptId() {
@@ -1026,6 +1048,18 @@
public URL getOutputLocation() {
return taskOutput;
}
+
+ public long getOffSet() {
+ return offSet;
+ }
+
+ public long getLength() {
+ return length;
+ }
+
+ public long getRawLength() {
+ return rawLength;
+ }
}
/** Describes the output of a map; could either be on disk or in-memory. */
@@ -1039,7 +1073,7 @@
byte[] data;
final boolean inMemory;
long compressedSize;
-
+ long offSet ;
public MapOutput(TaskID mapId, TaskAttemptID mapAttemptId,
Configuration conf, Path file, long size) {
this.mapId = mapId;
@@ -1047,6 +1081,7 @@
this.conf = conf;
this.file = file;
+ this.offSet = 0;
this.compressedSize = size;
this.data = null;
@@ -1054,6 +1089,21 @@
this.inMemory = false;
}
+ public MapOutput(TaskID mapId, TaskAttemptID mapAttemptId,
+ Configuration conf, Path file, long offSet, long length) {
+ this.mapId = mapId;
+ this.mapAttemptId = mapAttemptId;
+
+ this.conf = conf;
+ this.file = file;
+ this.offSet = offSet;
+ this.compressedSize = length;
+
+ this.data = null;
+
+ this.inMemory = false;
+ }
+
public MapOutput(TaskID mapId, TaskAttemptID mapAttemptId, byte[] data,
int compressedLength) {
this.mapId = mapId;
this.mapAttemptId = mapAttemptId;
@@ -1401,7 +1451,6 @@
// The size of the map-output
long bytes = mapOutput.compressedSize;
-
// lock the ReduceTask while we do the rename
synchronized (ReduceTask.this) {
if (copiedMapOutputs.contains(loc.getTaskId())) {
@@ -1411,12 +1460,12 @@
// Special case: discard empty map-outputs
if (bytes == 0) {
+ LOG.info("Discarding file as the size is zero " + mapOutput.file);
try {
mapOutput.discard();
} catch (IOException ioe) {
LOG.info("Couldn't discard output of " + loc.getTaskId());
}
-
// Note that we successfully copied the map-output
noteCopiedMapOutput(loc.getTaskId());
@@ -1439,8 +1488,11 @@
tmpMapOutput + " to " + filename);
}
- synchronized (mapOutputFilesOnDisk) {
- addToMapOutputFilesOnDisk(localFileSys.getFileStatus(filename));
+ synchronized (mapOutputFilesOnDisk) {
+ FileStatus fstatus = localFileSys.getFileStatus(filename);
+ fstatus.setLength(mapOutput.compressedSize);
+ fstatus.setOffset(mapOutput.offSet);
+ addToMapOutputFilesOnDisk(fstatus);
}
}
@@ -1476,12 +1528,47 @@
private MapOutput getMapOutput(MapOutputLocation mapOutputLoc,
Path filename, int reduce)
throws IOException, InterruptedException {
+
// Connect
URL url = mapOutputLoc.getOutputLocation();
URLConnection connection = url.openConnection();
- InputStream input = setupSecureConnection(mapOutputLoc, connection);
-
+ String tasktracker = url.getHost();
+ InputStream input = null;
+ MapOutput mapOutput = null;
+ if(tasktracker == null)
+ {
+ try{
+ if( ramManager.canFitInMemory(mapOutputLoc.getRawLength()) )
+ {
+ LOG.info("Task tracker is null , so doing shuffle to disk as
link but in memory");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Shuffling local map output " +
mapOutputLoc.getRawLength() + " bytes (" +
+ mapOutputLoc.getLength() + " raw bytes) " +
+ "into RAM from " + mapOutputLoc.getTaskAttemptId());
+ }
+ FileInputStream fis = new FileInputStream(new
File(mapOutputLoc.getOutputLocation().getPath()));
+ fis.getChannel().position(mapOutputLoc.getOffSet());
+ input = fis;
+ mapOutput = shuffleInMemory(mapOutputLoc, null, input,
+ (int)mapOutputLoc.getRawLength(),
+ (int)mapOutputLoc.getLength());
+ }
+ else
+ {
+ LOG.info("Task tracker is null , so doing shuffle to disk as
link");
+ mapOutput = shuffleToDiskAsLink(mapOutputLoc, filename);
+ }
+ return mapOutput;
+ } catch(IOException ioe)
+ {
+ LOG.error("Shuffle as Link did not work \n " + ioe.getMessage());
+ }
+ }
+ LOG.info("Task tracker is not null , so doing shuffle to disk usual
way");
+
+ input = setupSecureConnection(mapOutputLoc, connection);
+
// Validate header from map output
TaskAttemptID mapId = null;
try {
@@ -1534,14 +1621,12 @@
boolean shuffleInMemory =
ramManager.canFitInMemory(decompressedLength);
// Shuffle
- MapOutput mapOutput = null;
if (shuffleInMemory) {
if (LOG.isDebugEnabled()) {
LOG.debug("Shuffling " + decompressedLength + " bytes (" +
compressedLength + " raw bytes) " +
"into RAM from " + mapOutputLoc.getTaskAttemptId());
}
-
mapOutput = shuffleInMemory(mapOutputLoc, connection, input,
(int)decompressedLength,
(int)compressedLength);
@@ -1656,8 +1741,18 @@
if (!createdNow) {
// Reconnect
try {
- connection = mapOutputLoc.getOutputLocation().openConnection();
- input = setupSecureConnection(mapOutputLoc, connection);
+ URL url = mapOutputLoc.getOutputLocation();
+ if( connection == null)
+ {
+ FileInputStream fis = new FileInputStream(new
File(url.getPath()));
+ fis.getChannel().position(mapOutputLoc.getOffSet());
+ input = fis;
+ }
+ else
+ {
+ connection = url.openConnection();
+ input = setupSecureConnection(mapOutputLoc, connection);
+ }
} catch (IOException ioe) {
LOG.info("Failed reopen connection to fetch map-output from " +
mapOutputLoc.getHost());
@@ -1772,6 +1867,21 @@
return mapOutput;
}
+ private MapOutput shuffleToDiskAsLink(MapOutputLocation mapOutputLoc,
Path filename) throws IOException
+ {
+ // Find out a suitable location for the output on local-filesystem
+ Path localFilename =
+ lDirAlloc.getLocalPathForWrite(filename.toUri().getPath(),
+ mapOutputLoc.getLength(), conf);
+
+ MapOutput mapOutput =
+ new MapOutput(mapOutputLoc.getTaskId(),
mapOutputLoc.getTaskAttemptId(),
+ conf, localFileSys.makeQualified(localFilename),
+ mapOutputLoc.getOffSet(), mapOutputLoc.getLength());
+ FileUtil.symLink(mapOutputLoc.getOutputLocation().getPath(),
localFilename.toUri().getPath());
+ return mapOutput;
+ }
+
private MapOutput shuffleToDisk(MapOutputLocation mapOutputLoc,
InputStream input,
Path filename,
@@ -2352,6 +2462,7 @@
return false;
}
}
+ LOG.info("Completed FetchOutputs");
return mergeThrowable == null && copiedMapOutputs.size() == numMaps;
}
@@ -2433,6 +2544,7 @@
List<Segment<K,V>> memDiskSegments = new ArrayList<Segment<K,V>>();
long inMemToDiskBytes = 0;
if (mapOutputsFilesInMemory.size() > 0) {
+ LOG.info("In-memory files to be merged");
TaskID mapId = mapOutputsFilesInMemory.get(0).mapId;
inMemToDiskBytes = createInMemorySegments(memDiskSegments,
maxInMemReduce);
@@ -2449,16 +2561,15 @@
keyClass, valueClass, codec, null);
try {
Merger.writeFile(rIter, writer, reporter, job);
+ if (null != writer) {
+ writer.close();
+ }
addToMapOutputFilesOnDisk(fs.getFileStatus(outputPath));
} catch (Exception e) {
if (null != outputPath) {
fs.delete(outputPath, true);
}
throw new IOException("Final merge failed", e);
- } finally {
- if (null != writer) {
- writer.close();
- }
}
LOG.info("Merged " + numMemDiskSegments + " segments, " +
inMemToDiskBytes + " bytes to disk to satisfy " +
@@ -2472,15 +2583,15 @@
}
}
+ LOG.info("Completed in-memory merge and now moving to file merge");
// segments on disk
List<Segment<K,V>> diskSegments = new ArrayList<Segment<K,V>>();
long onDiskBytes = inMemToDiskBytes;
- Path[] onDisk = getMapFiles(fs, false);
- for (Path file : onDisk) {
- onDiskBytes += fs.getFileStatus(file).getLen();
- diskSegments.add(new Segment<K, V>(job, fs, file, codec, keepInputs));
+ for (FileStatus filestatus : mapOutputFilesOnDisk) {
+ onDiskBytes += filestatus.getLen();
+ diskSegments.add(new Segment<K, V>(job, fs, filestatus.getPath(),
filestatus.getOffset(), filestatus.getLen(), codec, keepInputs));
}
- LOG.info("Merging " + onDisk.length + " files, " +
+ LOG.info("Merging " + mapOutputFilesOnDisk.size() + " files, " +
onDiskBytes + " bytes from disk");
Collections.sort(diskSegments, new Comparator<Segment<K,V>>() {
public int compare(Segment<K, V> o1, Segment<K, V> o2) {
@@ -2604,9 +2715,10 @@
}
}
if(exitLocalFSMerge) {//to avoid running one extra time in the end
+ LOG.info("Exiting Local FS Merge");
break;
}
- List<Path> mapFiles = new ArrayList<Path>();
+ List<FileStatus> mapFiles = new ArrayList<FileStatus>();
long approxOutputSize = 0;
int bytesPerSum =
reduceTask.getConf().getInt("io.bytes.per.checksum", 512);
@@ -2620,16 +2732,19 @@
for (int i = 0; i < ioSortFactor; ++i) {
FileStatus filestatus = mapOutputFilesOnDisk.first();
mapOutputFilesOnDisk.remove(filestatus);
- mapFiles.add(filestatus.getPath());
+ mapFiles.add(filestatus);
approxOutputSize += filestatus.getLen();
+ LOG.info("File name:" + filestatus.getPath() +
+ " File Offset: " + filestatus.getOffset() +
+ " File Length: " + filestatus.getLength());
}
}
// sanity check
if (mapFiles.size() == 0) {
+ LOG.info("No map files to work with for FS Merger");
return;
}
-
// add the checksum length
approxOutputSize += ChecksumFileSystem
.getChecksumLength(approxOutputSize,
@@ -2651,11 +2766,11 @@
iter = Merger.merge(conf, rfs,
conf.getMapOutputKeyClass(),
conf.getMapOutputValueClass(),
- codec, mapFiles.toArray(new
Path[mapFiles.size()]),
+ codec, mapFiles.toArray(new
FileStatus[mapFiles.size()]),
true, ioSortFactor, tmpDir,
conf.getOutputKeyComparator(), reporter,
spilledRecordsCounter, null);
-
+
Merger.writeFile(iter, writer, reporter, conf);
writer.close();
} catch (Exception e) {
@@ -2842,14 +2957,15 @@
private int getMapCompletionEvents() throws IOException {
int numNewMaps = 0;
-
+ LOG.debug(" Received request for completionevents fromEventId " +
fromEventId.get() +
+ " Taskattemptid " + reduceTask.getTaskID() + " maxlocs " +
MAX_EVENTS_TO_FETCH );
MapTaskCompletionEventsUpdate update =
umbilical.getMapCompletionEvents(reduceTask.getJobID(),
fromEventId.get(),
MAX_EVENTS_TO_FETCH,
reduceTask.getTaskID(), jvmContext);
TaskCompletionEvent events[] = update.getMapTaskCompletionEvents();
-
+
// Check if the reset is required.
// Since there is no ordering of the task completion events at the
// reducer, the only option to sync with the new jobtracker is to
reset
@@ -2873,20 +2989,39 @@
switch (event.getTaskStatus()) {
case SUCCEEDED:
{
- URI u = URI.create(event.getTaskTrackerHttp());
- String host = u.getHost();
- TaskAttemptID taskId = event.getTaskAttemptId();
- URL mapOutputLocation = new URL(event.getTaskTrackerHttp() +
- "/mapOutput?job=" + taskId.getJobID() +
- "&map=" + taskId +
- "&reduce=" + getPartition());
+ URL mapOutputLocation = null;
+ String host = null;
+ TaskAttemptID taskId = null;
+ if(event.isSameTaskTrackerForMap()){
+
+ mapOutputLocation = new URL("file", null,
event.getMapOutputLocation());
+ host = "Local-FS";
+ taskId = event.getTaskAttemptId();
+ LOG.info("Map and Reducer are on the same machine: " +
+ " Offset " + event.getMapOutputOffSet() +
+ " Length " + event.getMapOutputReadLength() +
+ " Trackername " + event.getTaskTrackerName());
+
+ } else {
+ URI u = URI.create(event.getTaskTrackerHttp());
+ host = u.getHost();
+ taskId = event.getTaskAttemptId();
+ mapOutputLocation = new URL(event.getTaskTrackerHttp() +
+ "/mapOutput?job=" + taskId.getJobID() +
+ "&map=" + taskId +
+ "&reduce=" + getPartition());
+ LOG.info("Map and Reducer are not on the same machine: " +
+ " Offset " + event.getMapOutputOffSet() +
+ " Length " + event.getMapOutputReadLength() +
+ " Trackername " + event.getTaskTrackerName());
+ }
List<MapOutputLocation> loc = mapLocations.get(host);
if (loc == null) {
loc = Collections.synchronizedList
(new LinkedList<MapOutputLocation>());
mapLocations.put(host, loc);
}
- loc.add(new MapOutputLocation(taskId, host, mapOutputLocation));
+ loc.add(new MapOutputLocation(taskId, host,
mapOutputLocation,event.getMapOutputOffSet(),event.getMapOutputReadLength(),
event.getMapOutputRawLength()));
numNewMaps ++;
}
break;
Index: src/core/org/apache/hadoop/fs/FileStatus.java
===================================================================
--- src/core/org/apache/hadoop/fs/FileStatus.java (revision 1401127)
+++ src/core/org/apache/hadoop/fs/FileStatus.java (working copy)
@@ -39,10 +39,18 @@
private FsPermission permission;
private String owner;
private String group;
+ private long offset;
public FileStatus() { this(0, false, 0, 0, 0, 0, null, null, null, null); }
//We should deprecate this soon?
+ public FileStatus(long offset, long length, boolean isdir, int
block_replication,
+ long blocksize, long modification_time, Path path) {
+ this(length, isdir, block_replication, blocksize, modification_time,
+ 0, null, null, null, path);
+ this.offset = offset;
+ }
+
public FileStatus(long length, boolean isdir, int block_replication,
long blocksize, long modification_time, Path path) {
@@ -65,8 +73,25 @@
this.owner = (owner == null) ? "" : owner;
this.group = (group == null) ? "" : group;
this.path = path;
+ this.offset = 0;
}
+ public long getOffset() {
+ return offset;
+ }
+
+ public void setOffset(long offset) {
+ this.offset = offset;
+ }
+
+ public long getLength() {
+ return length;
+ }
+
+ public void setLength(long length) {
+ this.length = length;
+ }
+
/*
* @return the length of this file, in blocks
*/
@@ -191,6 +216,7 @@
permission.write(out);
Text.writeString(out, owner);
Text.writeString(out, group);
+ out.writeLong(offset);
}
public void readFields(DataInput in) throws IOException {
@@ -205,6 +231,7 @@
permission.readFields(in);
owner = Text.readString(in);
group = Text.readString(in);
+ offset = in.readLong();
}
/**
Index: src/mapred/org/apache/hadoop/mapred/TaskTracker.java
===================================================================
--- src/mapred/org/apache/hadoop/mapred/TaskTracker.java (revision
1401127)
+++ src/mapred/org/apache/hadoop/mapred/TaskTracker.java (working copy)
@@ -3525,10 +3525,15 @@
LOG.warn("Unknown child task fatalError: "+taskId+". Ignored.");
}
}
-
+
+ private static LRUCache<String, Path> fileCache = new LRUCache<String,
Path>(FILE_CACHE_SIZE);
+ private static LRUCache<String, Path> fileIndexCache = new LRUCache<String,
Path>(FILE_CACHE_SIZE);
+
public synchronized MapTaskCompletionEventsUpdate getMapCompletionEvents(
JobID jobId, int fromEventId, int maxLocs, TaskAttemptID id,
JvmContext jvmContext) throws IOException {
+ LOG.info(" Received request for completionevents fromEventId " +
fromEventId +
+ " Taskattemptid " + id + " maxlocs " + maxLocs );
TaskInProgress tip = runningTasks.get(id);
if (tip == null) {
throw new IOException("Unknown task; " + id
@@ -3542,7 +3547,10 @@
return new MapTaskCompletionEventsUpdate(mapEvents, true);
}
}
+
RunningJob rjob;
+ String userName = null;
+
synchronized (runningJobs) {
rjob = runningJobs.get(jobId);
if (rjob != null) {
@@ -3550,9 +3558,60 @@
FetchStatus f = rjob.getFetchStatus();
if (f != null) {
mapEvents = f.getMapEvents(fromEventId, maxLocs);
+ userName = rjob.jobConf.getUser();
+ }
+ }
+ }
+ }
+ for(TaskCompletionEvent event : mapEvents)
+ {
+ if( event.getTaskTrackerName().equals(this.taskTrackerName ) &&
event.getTaskStatus() == TaskCompletionEvent.Status.SUCCEEDED )
+ {
+ // Optimize
+ try{
+ event.setMapOutputLocal(true);
+ LOG.info(" Job id method " + jobId.toString() + " Job id event" +
event.getTaskAttemptId().getJobID().toString());
+ String intermediateOutputDir = getIntermediateOutputDir(userName,
event.getTaskAttemptId().getJobID().toString(),
event.getTaskAttemptId().toString());
+ // Index file
+ String indexKey = intermediateOutputDir + "/file.out.index";
+ Path indexFileName = fileIndexCache.get(indexKey);
+ if (indexFileName == null) {
+ indexFileName = lDirAlloc.getLocalPathToRead(indexKey, fConf);
+ fileIndexCache.put(indexKey, indexFileName);
+ }
+
+ // Map-output file
+ String fileKey = intermediateOutputDir + "/file.out";
+ Path mapOutputFileName = fileCache.get(fileKey);
+ if (mapOutputFileName == null) {
+ mapOutputFileName = lDirAlloc.getLocalPathToRead(fileKey, fConf);
+ fileCache.put(fileKey, mapOutputFileName);
}
+
+
+ String runAsUserName =
getTaskController().getRunAsUser(rjob.jobConf);
+ IndexRecord info =
+
indexCache.getIndexInformation(event.getTaskAttemptId().toString(),
tip.task.getPartition() ,indexFileName,
+ runAsUserName);
+
+ event.setMapOutputLocation(mapOutputFileName.toUri().getPath());
+ event.setMapOutputOffSet(info.startOffset);
+ event.setMapOutputReadLength(info.partLength);
+ event.setMapOutputRawLength(info.rawLength);
+ LOG.info(" Event info " + mapOutputFileName.toUri().getPath() +
+ " Offset " + info.startOffset +
+ " Length " + info.partLength +
+ " paritition " + tip.task.getPartition() +
+ " task attempt id " +
event.getTaskAttemptId().toString());
+ }
+ catch(Exception e)
+ {
+ // Continue as if nothing happened.
+ event.setMapOutputLocal(false);
}
}
+ else
+ LOG.info("TaskTracker local is " + this.taskTrackerName + " and the
event occured in " + event.getTaskTrackerName());
}
return new MapTaskCompletionEventsUpdate(mapEvents, false);
}
Index: src/mapred/org/apache/hadoop/mapred/Merger.java
===================================================================
--- src/mapred/org/apache/hadoop/mapred/Merger.java (revision 1401127)
+++ src/mapred/org/apache/hadoop/mapred/Merger.java (working copy)
@@ -28,6 +28,7 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.ChecksumFileSystem;
+import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.fs.Path;
@@ -66,6 +67,29 @@
public static <K extends Object, V extends Object>
RawKeyValueIterator merge(Configuration conf, FileSystem fs,
+ Class<K> keyClass, Class<V> valueClass,
+ CompressionCodec codec,
+ FileStatus[] inputs, boolean deleteInputs,
+ int mergeFactor, Path tmpDir,
+ RawComparator<K> comparator, Progressable reporter,
+ Counters.Counter readsCounter,
+ Counters.Counter writesCounter)
+ throws IOException {
+ // segments on disk
+ List<Segment<K,V>> diskSegments = new ArrayList<Segment<K,V>>();
+ for (FileStatus filestatus : inputs) {
+ diskSegments.add(new Segment<K, V>(conf, fs, filestatus.getPath(),
filestatus.getOffset(), filestatus.getLen(), codec, !deleteInputs));
+ }
+
+ return new MergeQueue<K, V>(conf, fs, diskSegments, comparator, reporter,
+ true).merge(keyClass, valueClass,
+ mergeFactor,
+ tmpDir,
+ readsCounter, writesCounter);
+ }
+
+ public static <K extends Object, V extends Object>
+ RawKeyValueIterator merge(Configuration conf, FileSystem fs,
Class<K> keyClass, Class<V> valueClass,
CompressionCodec codec,
List<Segment<K, V>> segments,
@@ -179,7 +203,7 @@
CompressionCodec codec, boolean preserve) throws
IOException {
this(conf, fs, file, 0, fs.getFileStatus(file).getLen(), codec,
preserve);
}
-
+
public Segment(Configuration conf, FileSystem fs, Path file,
long segmentOffset, long segmentLength, CompressionCodec codec,
boolean preserve) throws IOException {
@@ -192,7 +216,7 @@
this.segmentOffset = segmentOffset;
this.segmentLength = segmentLength;
}
-
+
public Segment(Reader<K, V> reader, boolean preserve) {
this.reader = reader;
this.preserve = preserve;
@@ -204,7 +228,7 @@
if (reader == null) {
FSDataInputStream in = fs.open(file);
in.seek(segmentOffset);
- reader = new Reader<K, V>(conf, in, segmentLength, codec,
readsCounter);
+ reader = new Reader<K, V>(conf, in, segmentLength, codec,
readsCounter);
}
}
