You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ss...@apache.org on 2015/09/22 20:04:53 UTC
[1/2] tez git commit: TEZ_2844. Backport TEZ-2775 to branch-0.7.
Improve and consolidate logging in Runtime components. (sseth)
Repository: tez
Updated Branches:
refs/heads/branch-0.7 c7a946c9c -> cd80d9ab5
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/Shuffle.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/Shuffle.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/Shuffle.java
index 1eca9e9..417e66a 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/Shuffle.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/Shuffle.java
@@ -188,7 +188,7 @@ public class Shuffle implements ExceptionReporter {
TezCounter bytesShuffedToMem = inputContext.getCounters().findCounter(
TaskCounter.SHUFFLE_BYTES_TO_MEM);
- LOG.info("Shuffle assigned with " + numInputs + " inputs" + ", codec: "
+ LOG.info(srcNameTrimmed + ": " + "Shuffle assigned with " + numInputs + " inputs" + ", codec: "
+ (codec == null ? "None" : codec.getClass().getName()) +
"ifileReadAhead: " + ifileReadAhead);
@@ -207,7 +207,8 @@ public class Shuffle implements ExceptionReporter {
bytesShuffedToDisk,
bytesShuffedToDiskDirect,
bytesShuffedToMem,
- startTime);
+ startTime,
+ srcNameTrimmed);
this.mergePhaseTime = inputContext.getCounters().findCounter(TaskCounter.MERGE_PHASE_TIME);
this.shufflePhaseTime = inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_PHASE_TIME);
@@ -232,14 +233,14 @@ public class Shuffle implements ExceptionReporter {
sslShuffle);
ExecutorService rawExecutor = Executors.newFixedThreadPool(1, new ThreadFactoryBuilder()
- .setDaemon(true).setNameFormat("ShuffleAndMergeRunner [" + srcNameTrimmed + "]").build());
+ .setDaemon(true).setNameFormat("ShuffleAndMergeRunner {" + srcNameTrimmed + "}").build());
int configuredNumFetchers =
conf.getInt(
TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES,
TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES_DEFAULT);
numFetchers = Math.min(configuredNumFetchers, numInputs);
- LOG.info("Num fetchers being started: " + numFetchers);
+ LOG.info(srcNameTrimmed + ": " + "Num fetchers being started: " + numFetchers);
fetchers = Lists.newArrayListWithCapacity(numFetchers);
localDiskFetchEnabled = conf.getBoolean(
TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH,
@@ -253,7 +254,7 @@ public class Shuffle implements ExceptionReporter {
if (!isShutDown.get()) {
eventHandler.handleEvents(events);
} else {
- LOG.info("Ignoring events since already shutdown. EventCount: " + events.size());
+ LOG.info(srcNameTrimmed + ": " + "Ignoring events since already shutdown. EventCount: " + events.size());
}
}
@@ -387,7 +388,7 @@ public class Shuffle implements ExceptionReporter {
}
inputContext.inputIsReady();
- LOG.info("merge complete for input vertex : " + inputContext.getSourceVertexName());
+ LOG.info("merge complete for input vertex : " + srcNameTrimmed);
return kvIter;
}
}
@@ -400,16 +401,16 @@ public class Shuffle implements ExceptionReporter {
for (FetcherOrderedGrouped fetcher : fetchers) {
try {
fetcher.shutDown();
- LOG.info("Shutdown.." + fetcher.getName() + ", status:" + fetcher.isAlive() + ", "
+ LOG.info(srcNameTrimmed + ": " + "Shutdown.." + fetcher.getName() + ", status:" + fetcher.isAlive() + ", "
+ "isInterrupted:" + fetcher.isInterrupted());
} catch (InterruptedException e) {
if (ignoreErrors) {
- LOG.info("Interrupted while shutting down fetchers. Ignoring.");
+ LOG.info(srcNameTrimmed + ": " + "Interrupted while shutting down fetchers. Ignoring.");
} else {
if (ie != null) {
ie = e;
} else {
- LOG.warn(
+ LOG.warn(srcNameTrimmed + ": " +
"Ignoring exception while shutting down fetcher since a previous one was seen and will be thrown "
+ e);
}
@@ -432,7 +433,7 @@ public class Shuffle implements ExceptionReporter {
scheduler.close();
} catch (InterruptedException e) {
if (ignoreErrors) {
- LOG.info("Interrupted while attempting to close the scheduler during cleanup. Ignoring");
+ LOG.info(srcNameTrimmed + ": " + "Interrupted while attempting to close the scheduler during cleanup. Ignoring");
} else {
throw e;
}
@@ -446,7 +447,7 @@ public class Shuffle implements ExceptionReporter {
merger.close();
} catch (Throwable e) {
if (ignoreErrors) {
- LOG.info("Exception while trying to shutdown merger, Ignoring", e);
+ LOG.info(srcNameTrimmed + ": " + "Exception while trying to shutdown merger, Ignoring", e);
} else {
throw e;
}
@@ -456,11 +457,14 @@ public class Shuffle implements ExceptionReporter {
private void cleanupIgnoreErrors() {
try {
+ if (eventHandler != null) {
+ eventHandler.logProgress(true);
+ }
cleanupFetchers(true);
cleanupShuffleScheduler(true);
cleanupMerger(true);
} catch (Throwable t) {
- LOG.info("Error in cleaning up.., ", t);
+ LOG.info(srcNameTrimmed + ": " + "Error in cleaning up.., ", t);
}
}
@@ -468,6 +472,8 @@ public class Shuffle implements ExceptionReporter {
public synchronized void reportException(Throwable t) {
// RunShuffleCallable onFailure deals with ignoring errors on shutdown.
if (throwable.get() == null) {
+ LOG.info(srcNameTrimmed + ": " + "Setting throwable in reportException with message [" + t.getMessage() +
+ "] from thread [" + Thread.currentThread().getName());
throwable.set(t);
throwingThreadName = Thread.currentThread().getName();
// Notify the scheduler so that the reporting thread finds the
@@ -494,15 +500,15 @@ public class Shuffle implements ExceptionReporter {
private class ShuffleRunnerFutureCallback implements FutureCallback<TezRawKeyValueIterator> {
@Override
public void onSuccess(TezRawKeyValueIterator result) {
- LOG.info("Shuffle Runner thread complete");
+ LOG.info(srcNameTrimmed + ": " + "Shuffle Runner thread complete");
}
@Override
public void onFailure(Throwable t) {
if (isShutDown.get()) {
- LOG.info("Already shutdown. Ignoring error: ", t);
+ LOG.info(srcNameTrimmed + ": " + "Already shutdown. Ignoring error");
} else {
- LOG.error("ShuffleRunner failed with error", t);
+ LOG.error(srcNameTrimmed + ": " + "ShuffleRunner failed with error", t);
inputContext.fatalError(t, "Shuffle Runner Failed");
cleanupIgnoreErrors();
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleInputEventHandlerOrderedGrouped.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleInputEventHandlerOrderedGrouped.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleInputEventHandlerOrderedGrouped.java
index 32ac766..8d0c357 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleInputEventHandlerOrderedGrouped.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleInputEventHandlerOrderedGrouped.java
@@ -22,9 +22,11 @@ import java.io.IOException;
import java.net.URI;
import java.util.BitSet;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.annotations.VisibleForTesting;
import com.google.protobuf.ByteString;
+import org.apache.tez.runtime.library.common.shuffle.ShuffleEventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.tez.common.TezCommonUtils;
@@ -41,7 +43,7 @@ import org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads.DataMovem
import com.google.protobuf.InvalidProtocolBufferException;
-public class ShuffleInputEventHandlerOrderedGrouped {
+public class ShuffleInputEventHandlerOrderedGrouped implements ShuffleEventHandler {
private static final Logger LOG = LoggerFactory.getLogger(ShuffleInputEventHandlerOrderedGrouped.class);
@@ -51,6 +53,11 @@ public class ShuffleInputEventHandlerOrderedGrouped {
private int maxMapRuntime = 0;
private final boolean sslShuffle;
+ private final AtomicInteger nextToLogEventCount = new AtomicInteger(0);
+ private final AtomicInteger numDmeEvents = new AtomicInteger(0);
+ private final AtomicInteger numObsoletionEvents = new AtomicInteger(0);
+ private final AtomicInteger numDmeEventsNoData = new AtomicInteger(0);
+
public ShuffleInputEventHandlerOrderedGrouped(InputContext inputContext,
ShuffleScheduler scheduler, boolean sslShuffle) {
this.inputContext = inputContext;
@@ -58,20 +65,36 @@ public class ShuffleInputEventHandlerOrderedGrouped {
this.sslShuffle = sslShuffle;
}
+ @Override
public void handleEvents(List<Event> events) throws IOException {
for (Event event : events) {
handleEvent(event);
}
}
-
-
+
+ @Override
+ public void logProgress(boolean updateOnClose) {
+ LOG.info(inputContext.getSourceVertexName() + ": "
+ + "numDmeEventsSeen=" + numDmeEvents.get()
+ + ", numDmeEventsSeenWithNoData=" + numDmeEventsNoData.get()
+ + ", numObsoletionEventsSeen=" + numObsoletionEvents.get()
+ + (updateOnClose == true ? ", updateOnClose" : ""));
+ }
+
private void handleEvent(Event event) throws IOException {
if (event instanceof DataMovementEvent) {
+ numDmeEvents.incrementAndGet();
processDataMovementEvent((DataMovementEvent) event);
scheduler.updateEventReceivedTime();
} else if (event instanceof InputFailedEvent) {
+ numObsoletionEvents.incrementAndGet();
processTaskFailedEvent((InputFailedEvent) event);
}
+ if (numDmeEvents.get() + numObsoletionEvents.get() > nextToLogEventCount.get()) {
+ logProgress(false);
+ // Log every 50 events seen.
+ nextToLogEventCount.addAndGet(50);
+ }
}
private void processDataMovementEvent(DataMovementEvent dmEvent) throws IOException {
@@ -82,8 +105,11 @@ public class ShuffleInputEventHandlerOrderedGrouped {
throw new TezUncheckedException("Unable to parse DataMovementEvent payload", e);
}
int partitionId = dmEvent.getSourceIndex();
- LOG.info("DME srcIdx: " + partitionId + ", targetIdx: " + dmEvent.getTargetIndex()
- + ", attemptNum: " + dmEvent.getVersion() + ", payload: " + ShuffleUtils.stringify(shufflePayload));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("DME srcIdx: " + partitionId + ", targetIdx: " + dmEvent.getTargetIndex()
+ + ", attemptNum: " + dmEvent.getVersion() + ", payload: " +
+ ShuffleUtils.stringify(shufflePayload));
+ }
// TODO NEWTEZ See if this duration hack can be removed.
int duration = shufflePayload.getRunDuration();
if (duration > maxMapRuntime) {
@@ -101,6 +127,7 @@ public class ShuffleInputEventHandlerOrderedGrouped {
"Source partition: " + partitionId + " did not generate any data. SrcAttempt: ["
+ srcAttemptIdentifier + "]. Not fetching.");
}
+ numDmeEventsNoData.incrementAndGet();
scheduler.copySucceeded(srcAttemptIdentifier, null, 0, 0, 0, null);
return;
}
@@ -120,10 +147,11 @@ public class ShuffleInputEventHandlerOrderedGrouped {
private void processTaskFailedEvent(InputFailedEvent ifEvent) {
InputAttemptIdentifier taIdentifier = new InputAttemptIdentifier(ifEvent.getTargetIndex(), ifEvent.getVersion());
scheduler.obsoleteInput(taIdentifier);
- LOG.info("Obsoleting output of src-task: " + taIdentifier);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Obsoleting output of src-task: " + taIdentifier);
+ }
}
- // TODO NEWTEZ Handle encrypted shuffle
@VisibleForTesting
URI getBaseURI(String host, int port, int partitionId) {
StringBuilder sb = ShuffleUtils.constructBaseURIForShuffleHandler(host, port,
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java
index cdfd511..b98bc56 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java
@@ -33,6 +33,7 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
@@ -71,6 +72,7 @@ class ShuffleScheduler {
private boolean[] finishedMaps;
private final int numInputs;
+ private final String srcNameTrimmed;
private int remainingMaps;
private int numFetchedSpills;
private Map<String, MapHost> mapLocations = new HashMap<String, MapHost>();
@@ -128,12 +130,14 @@ class ShuffleScheduler {
TezCounter failedShuffleCounter,
TezCounter bytesShuffledToDisk,
TezCounter bytesShuffledToDiskDirect,
- TezCounter bytesShuffledToMem, long startTime) {
+ TezCounter bytesShuffledToMem, long startTime,
+ String srcNameTrimmed) {
this.inputContext = inputContext;
this.numInputs = numberOfInputs;
abortFailureLimit = Math.max(30, numberOfInputs / 10);
remainingMaps = numberOfInputs;
finishedMaps = new boolean[remainingMaps]; // default init to false
+ this.srcNameTrimmed = srcNameTrimmed;
this.referee = new Referee();
this.shuffle = shuffle;
this.shuffledInputsCounter = shuffledInputsCounter;
@@ -312,7 +316,7 @@ class ShuffleScheduler {
}
if (remainingMaps == 0) {
- LOG.info("All inputs fetched for input vertex : " + inputContext.getSourceVertexName());
+ LOG.info(srcNameTrimmed + ": " + "All inputs fetched for input vertex : " + inputContext.getSourceVertexName());
notifyAll();
}
@@ -330,7 +334,7 @@ class ShuffleScheduler {
}
} else {
// input is already finished. duplicate fetch.
- LOG.warn("Duplicate fetch of input no longer needs to be fetched: " + srcAttemptIdentifier);
+ LOG.warn(srcNameTrimmed + ": " + "Duplicate fetch of input no longer needs to be fetched: " + srcAttemptIdentifier);
// free the resource - specially memory
// If the src does not generate data, output will be null.
@@ -364,19 +368,24 @@ class ShuffleScheduler {
@VisibleForTesting
void reportExceptionForInput(Exception exception) {
- LOG.error("Reporting exception for input", exception);
+ LOG.error(srcNameTrimmed + ": " + "Reporting exception for input", exception);
shuffle.reportException(exception);
}
+ private final AtomicInteger nextProgressLineEventCount = new AtomicInteger(0);
+
private void logProgress() {
- double mbs = (double) totalBytesShuffledTillNow / (1024 * 1024);
int inputsDone = numInputs - remainingMaps;
- long secsSinceStart = (System.currentTimeMillis() - startTime) / 1000 + 1;
-
- double transferRate = mbs / secsSinceStart;
- LOG.info("copy(" + inputsDone + " (spillsFetched=" + numFetchedSpills + ") of " + numInputs +
- ". Transfer rate (CumulativeDataFetched/TimeSinceInputStarted)) "
- + mbpsFormat.format(transferRate) + " MB/s)");
+ if (inputsDone > nextProgressLineEventCount.get() || inputsDone == numInputs) {
+ nextProgressLineEventCount.addAndGet(50);
+ double mbs = (double) totalBytesShuffledTillNow / (1024 * 1024);
+ long secsSinceStart = (System.currentTimeMillis() - startTime) / 1000 + 1;
+
+ double transferRate = mbs / secsSinceStart;
+ LOG.info("copy(" + inputsDone + " (spillsFetched=" + numFetchedSpills + ") of " + numInputs +
+ ". Transfer rate (CumulativeDataFetched/TimeSinceInputStarted)) "
+ + mbpsFormat.format(transferRate) + " MB/s)");
+ }
}
public synchronized void copyFailed(InputAttemptIdentifier srcAttempt,
@@ -431,7 +440,7 @@ class ShuffleScheduler {
}
public void reportLocalError(IOException ioe) {
- LOG.error("Shuffle failed : caused by local error", ioe);
+ LOG.error(srcNameTrimmed + ": " + "Shuffle failed : caused by local error", ioe);
// Shuffle knows how to deal with failures post shutdown via the onFailure hook
shuffle.reportException(ioe);
}
@@ -444,7 +453,7 @@ class ShuffleScheduler {
boolean connectError) {
if ((reportReadErrorImmediately && (readError || connectError))
|| ((failures % maxFetchFailuresBeforeReporting) == 0)) {
- LOG.info("Reporting fetch failure for InputIdentifier: "
+ LOG.info(srcNameTrimmed + ": " + "Reporting fetch failure for InputIdentifier: "
+ srcAttempt + " taskAttemptIdentifier: "
+ TezRuntimeUtils.getTaskAttemptIdentifier(
inputContext.getSourceVertexName(), srcAttempt.getInputIdentifier().getInputIndex(),
@@ -501,7 +510,8 @@ class ShuffleScheduler {
failureCounts.size() == (numInputs - doneMaps))
&& !reducerHealthy
&& (!reducerProgressedEnough || reducerStalled)) {
- LOG.error("Shuffle failed with too many fetch failures " + "and insufficient progress!"
+ LOG.error(srcNameTrimmed + ": " + "Shuffle failed with too many fetch failures " +
+ "and insufficient progress!"
+ "failureCounts=" + failureCounts.size() + ", pendingInputs=" + (numInputs - doneMaps)
+ ", reducerHealthy=" + reducerHealthy + ", reducerProgressedEnough="
+ reducerProgressedEnough + ", reducerStalled=" + reducerStalled);
@@ -546,7 +556,7 @@ class ShuffleScheduler {
public synchronized void obsoleteInput(InputAttemptIdentifier srcAttempt) {
// The incoming srcAttempt does not contain a path component.
- LOG.info("Adding obsolete input: " + srcAttempt);
+ LOG.info(srcNameTrimmed + ": " + "Adding obsolete input: " + srcAttempt);
if (shuffleInfoEventsMap.containsKey(srcAttempt.getInputIdentifier())) {
//Pipelined shuffle case (where shuffleInfoEventsMap gets populated).
//Fail fast here.
@@ -565,6 +575,9 @@ class ShuffleScheduler {
public synchronized MapHost getHost() throws InterruptedException {
while(pendingHosts.isEmpty()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("PendingHosts=" + pendingHosts);
+ }
wait();
}
@@ -578,7 +591,7 @@ class ShuffleScheduler {
pendingHosts.remove(host);
host.markBusy();
if (LOG.isDebugEnabled()) {
- LOG.debug("Assigning " + host + " with " + host.getNumKnownMapOutputs() +
+ LOG.debug(srcNameTrimmed + ": " + "Assigning " + host + " with " + host.getNumKnownMapOutputs() +
" to " + Thread.currentThread().getName());
}
shuffleStart.set(System.currentTimeMillis());
@@ -685,8 +698,10 @@ class ShuffleScheduler {
notifyAll();
}
}
- LOG.info(host + " freed by " + Thread.currentThread().getName() + " in " +
- (System.currentTimeMillis()-shuffleStart.get()) + "ms");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(host + " freed by " + Thread.currentThread().getName() + " in " +
+ (System.currentTimeMillis() - shuffleStart.get()) + "ms");
+ }
}
public synchronized void resetKnownMaps() {
@@ -752,8 +767,8 @@ class ShuffleScheduler {
*/
private class Referee extends Thread {
public Referee() {
- setName("ShufflePenaltyReferee ["
- + TezUtilsInternal.cleanVertexName(inputContext.getSourceVertexName()) + "]");
+ setName("ShufflePenaltyReferee {"
+ + TezUtilsInternal.cleanVertexName(inputContext.getSourceVertexName()) + "}");
setDaemon(true);
}
@@ -780,6 +795,7 @@ class ShuffleScheduler {
}
public void close() throws InterruptedException {
+ logProgress();
referee.interrupt();
referee.join();
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/ExternalSorter.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/ExternalSorter.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/ExternalSorter.java
index fd62cfa..aba04e0 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/ExternalSorter.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/ExternalSorter.java
@@ -155,7 +155,10 @@ public abstract class ExternalSorter {
rfs = ((LocalFileSystem)FileSystem.getLocal(this.conf)).getRaw();
- LOG.info("Initial Mem : " + initialMemoryAvailable + ", assignedMb=" + ((initialMemoryAvailable >> 20)));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(outputContext.getDestinationVertexName() + ": Initial Mem bytes : " +
+ initialMemoryAvailable + ", in MB=" + ((initialMemoryAvailable >> 20)));
+ }
int assignedMb = (int) (initialMemoryAvailable >> 20);
//Let the overflow checks happen in appropriate sorter impls
this.availableMemoryMb = assignedMb;
@@ -173,9 +176,13 @@ public abstract class ExternalSorter {
serializationFactory = new SerializationFactory(this.conf);
keySerializer = serializationFactory.getSerializer(keyClass);
valSerializer = serializationFactory.getSerializer(valClass);
- LOG.info("keySerializer=" + keySerializer + "; valueSerializer=" + valSerializer
- + "; comparator=" + (RawComparator) ConfigUtils.getIntermediateOutputKeyComparator(conf)
- + "; conf=" + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
+ LOG.info(outputContext.getDestinationVertexName() + " using: "
+ + "memoryMb=" + assignedMb
+ + ", keySerializerClass=" + keyClass
+ + ", valueSerializerClass=" + valSerializer
+ + ", comparator=" + (RawComparator) ConfigUtils.getIntermediateOutputKeyComparator(conf)
+ + ", partitioner=" + conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS)
+ + ", serialization=" + conf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY));
// counters
mapOutputByteCounter = outputContext.getCounters().findCounter(TaskCounter.OUTPUT_BYTES);
@@ -230,8 +237,7 @@ public abstract class ExternalSorter {
// Task outputs
mapOutputFile = TezRuntimeUtils.instantiateTaskOutputManager(conf, outputContext);
-
- LOG.info("Instantiating Partitioner: [" + conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS) + "]");
+
this.conf.setInt(TezRuntimeFrameworkConfigs.TEZ_RUNTIME_NUM_EXPECTED_PARTITIONS, this.partitions);
this.partitioner = TezRuntimeUtils.instantiatePartitioner(this.conf);
this.combiner = TezRuntimeUtils.instantiateCombiner(this.conf, outputContext);
@@ -322,9 +328,11 @@ public abstract class ExternalSorter {
TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB + " " + initialMemRequestMb + " should be "
+ "larger than 0 and should be less than the available task memory (MB):" +
(maxAvailableTaskMemory >> 20));
- LOG.info("Requested SortBufferSize ("
- + TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB + "): "
- + initialMemRequestMb);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Requested SortBufferSize ("
+ + TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB + "): "
+ + initialMemRequestMb);
+ }
return reqBytes;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java
index fe96688..049087b 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java
@@ -1,4 +1,4 @@
-/**
+ /**
* 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
@@ -119,7 +119,9 @@ public class PipelinedSorter extends ExternalSorter {
PipelinedSorter(OutputContext outputContext, Configuration conf, int numOutputs,
long initialMemoryAvailable, int blkSize) throws IOException {
super(outputContext, conf, numOutputs, initialMemoryAvailable);
-
+
+ StringBuilder initialSetupLogLine = new StringBuilder("Setting up PipelinedSorter for ")
+ .append(outputContext.getDestinationVertexName()).append(": ");
partitionBits = bitcount(partitions)+1;
boolean confPipelinedShuffle = this.conf.getBoolean(TezRuntimeConfiguration
@@ -139,10 +141,29 @@ public class PipelinedSorter extends ExternalSorter {
long usage = sortmb << 20;
//Divide total memory into different blocks.
int numberOfBlocks = Math.max(1, (int) Math.ceil(1.0 * usage / blockSize));
- LOG.info("Number of Blocks : " + numberOfBlocks
- + ", maxMemUsage=" + maxMemUsage + ", BLOCK_SIZE=" + blockSize + ", finalMergeEnabled="
- + isFinalMergeEnabled() + ", pipelinedShuffle=" + pipelinedShuffle + ", "
- + "sendEmptyPartitionDetails=" + sendEmptyPartitionDetails);
+ initialSetupLogLine.append("#blocks=").append(numberOfBlocks);
+ initialSetupLogLine.append(", maxMemUsage=").append(maxMemUsage);
+ initialSetupLogLine.append(", BLOCK_SIZE=").append(blockSize);
+ initialSetupLogLine.append(", finalMergeEnabled=").append(isFinalMergeEnabled());
+ initialSetupLogLine.append(", pipelinedShuffle=").append(pipelinedShuffle);
+ initialSetupLogLine.append(", sendEmptyPartitions=").append(sendEmptyPartitionDetails);
+ initialSetupLogLine.append(", ").append(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB).append(
+ "=").append(
+ sortmb);
+
+
+ initialSetupLogLine.append(", UsingHashComparator=");
+ // k/v serialization
+ if(comparator instanceof ProxyComparator) {
+ hasher = (ProxyComparator)comparator;
+ initialSetupLogLine.append(true);
+ } else {
+ hasher = null;
+ initialSetupLogLine.append(false);
+ }
+
+ LOG.info(initialSetupLogLine.toString());
+
long totalCapacityWithoutMeta = 0;
for (int i = 0; i < numberOfBlocks; i++) {
Preconditions.checkArgument(usage > 0, "usage can't be less than zero " + usage);
@@ -156,7 +177,6 @@ public class PipelinedSorter extends ExternalSorter {
listIterator = bufferList.listIterator();
- LOG.info(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB + " = " + sortmb);
Preconditions.checkArgument(listIterator.hasNext(), "Buffer list seems to be empty " + bufferList.size());
span = new SortSpan(listIterator.next(), 1024*1024, 16, this.comparator);
merger = new SpanMerger(); // SpanIterators are comparable
@@ -166,17 +186,11 @@ public class PipelinedSorter extends ExternalSorter {
TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SORTER_SORT_THREADS_DEFAULT);
sortmaster = Executors.newFixedThreadPool(sortThreads,
new ThreadFactoryBuilder().setDaemon(true)
- .setNameFormat("Sorter [" + TezUtilsInternal
- .cleanVertexName(outputContext.getDestinationVertexName()) + "] #%d")
+ .setNameFormat("Sorter {" + TezUtilsInternal
+ .cleanVertexName(outputContext.getDestinationVertexName()) + "} #%d")
.build());
- // k/v serialization
- if(comparator instanceof ProxyComparator) {
- hasher = (ProxyComparator)comparator;
- LOG.info("Using the HashComparator");
- } else {
- hasher = null;
- }
+
valSerializer.open(span.out);
keySerializer.open(span.out);
minSpillsForCombine = this.conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINE_MIN_SPILLS, 3);
@@ -217,6 +231,9 @@ public class PipelinedSorter extends ExternalSorter {
spill();
stopWatch.stop();
LOG.info("Time taken for spill " + (stopWatch.elapsedMillis()) + " ms");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(outputContext.getDestinationVertexName() + ": Time taken for spill " + (stopWatch.elapsedMillis()) + " ms");
+ }
if (pipelinedShuffle) {
sendPipelinedShuffleEvents();
}
@@ -256,7 +273,8 @@ public class PipelinedSorter extends ExternalSorter {
(numSpills - 1), indexCacheList.get(numSpills - 1), partitions, sendEmptyPartitionDetails,
pathComponent);
outputContext.sendEvents(events);
- LOG.info("Added spill event for spill (final update=false), spillId=" + (numSpills - 1));
+ LOG.info(outputContext.getDestinationVertexName() +
+ ": Added spill event for spill (final update=false), spillId=" + (numSpills - 1));
}
@Override
@@ -362,11 +380,15 @@ public class PipelinedSorter extends ExternalSorter {
final TezSpillRecord spillRec = new TezSpillRecord(partitions);
// getSpillFileForWrite with size -1 as the serialized size of KV pair is still unknown
final Path filename = mapOutputFile.getSpillFileForWrite(numSpills, -1);
+ Path indexFilename =
+ mapOutputFile.getSpillIndexFileForWrite(numSpills, partitions
+ * MAP_OUTPUT_INDEX_RECORD_LENGTH);
spillFilePaths.put(numSpills, filename);
FSDataOutputStream out = rfs.create(filename, true, 4096);
try {
- LOG.info("Spilling to " + filename.toString());
+ LOG.info(outputContext.getDestinationVertexName() + ": Spilling to " + filename.toString() +
+ ", indexFilename=" + indexFilename);
for (int i = 0; i < partitions; ++i) {
Writer writer = null;
try {
@@ -399,10 +421,6 @@ public class PipelinedSorter extends ExternalSorter {
}
}
- Path indexFilename =
- mapOutputFile.getSpillIndexFileForWrite(numSpills, partitions
- * MAP_OUTPUT_INDEX_RECORD_LENGTH);
- LOG.info("Spill Index filename:" + indexFilename);
spillFileIndexPaths.put(numSpills, indexFilename);
spillRec.writeToFile(indexFilename, conf);
//TODO: honor cache limits
@@ -433,7 +451,7 @@ public class PipelinedSorter extends ExternalSorter {
try {
merger.ready(); // wait for all the future results from sort threads
- LOG.info("Spilling to " + filename.toString());
+ LOG.info(outputContext.getDestinationVertexName() + ": Spilling to " + filename.toString());
for (int i = 0; i < partitions; ++i) {
TezRawKeyValueIterator kvIter = merger.filter(i);
//write merged output to disk
@@ -484,7 +502,7 @@ public class PipelinedSorter extends ExternalSorter {
public void flush() throws IOException {
final String uniqueIdentifier = outputContext.getUniqueIdentifier();
- LOG.info("Starting flush of map output");
+ LOG.info(outputContext.getDestinationVertexName() + ": Starting flush of map output");
span.end();
merger.add(span.sort(sorter));
spill();
@@ -495,7 +513,14 @@ public class PipelinedSorter extends ExternalSorter {
if(indexCacheList.isEmpty()) {
- LOG.warn("Index list is empty... returning");
+ /*
+ * If we do not have this check, and if the task gets killed in the middle, it can throw
+ * NPE leading to distraction when debugging.
+ */
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(outputContext.getDestinationVertexName()
+ + ": Index list is empty... returning");
+ }
return;
}
@@ -514,7 +539,7 @@ public class PipelinedSorter extends ExternalSorter {
ShuffleUtils.generateEventOnSpill(events, isFinalMergeEnabled(), isLastEvent,
outputContext, i, indexCacheList.get(i), partitions,
sendEmptyPartitionDetails, pathComponent);
- LOG.info("Adding spill event for spill (final update=" + isLastEvent + "), spillId=" + i);
+ LOG.info(outputContext.getDestinationVertexName() + ": Adding spill event for spill (final update=" + isLastEvent + "), spillId=" + i);
}
outputContext.sendEvents(events);
return;
@@ -533,8 +558,9 @@ public class PipelinedSorter extends ExternalSorter {
sameVolRename(filename, finalOutputFile);
sameVolRename(indexFilename, finalIndexFile);
- if (LOG.isInfoEnabled()) {
- LOG.info("numSpills=" + numSpills + ", finalOutputFile=" + finalOutputFile + ", "
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(outputContext.getDestinationVertexName() + ": numSpills=" + numSpills +
+ ", finalOutputFile=" + finalOutputFile + ", "
+ "finalIndexFile=" + finalIndexFile + ", filename=" + filename + ", indexFilename=" +
indexFilename);
}
@@ -549,7 +575,8 @@ public class PipelinedSorter extends ExternalSorter {
mapOutputFile.getOutputIndexFileForWrite(0); //TODO
if (LOG.isDebugEnabled()) {
- LOG.debug("numSpills: " + numSpills + ", finalOutputFile:" + finalOutputFile + ", finalIndexFile:"
+ LOG.debug(outputContext.getDestinationVertexName() + ": " +
+ "numSpills: " + numSpills + ", finalOutputFile:" + finalOutputFile + ", finalIndexFile:"
+ finalIndexFile);
}
@@ -705,8 +732,8 @@ public class PipelinedSorter extends ExternalSorter {
}
ByteBuffer reserved = source.duplicate();
reserved.mark();
- LOG.info("reserved.remaining() = " + reserved.remaining());
- LOG.info("reserved.metasize = "+ metasize);
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "reserved.remaining()=" +
+ reserved.remaining() + ", reserved.metasize=" + metasize);
reserved.position(metasize);
kvbuffer = reserved.slice();
reserved.flip();
@@ -725,7 +752,7 @@ public class PipelinedSorter extends ExternalSorter {
if(length() > 1) {
sorter.sort(this, 0, length(), nullProgressable);
}
- LOG.info("done sorting span=" + index + ", length=" + length() + ", "
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "done sorting span=" + index + ", length=" + length() + ", "
+ "time=" + (System.currentTimeMillis() - start));
return new SpanIterator(this);
}
@@ -793,7 +820,7 @@ public class PipelinedSorter extends ExternalSorter {
newSpan = new SortSpan(remaining, items, perItem,
ConfigUtils.getIntermediateOutputKeyComparator(conf));
newSpan.index = index+1;
- LOG.info(String.format("New Span%d.length = %d, perItem = %d", newSpan.index, newSpan
+ LOG.info(String.format(outputContext.getDestinationVertexName() + ": " + "New Span%d.length = %d, perItem = %d", newSpan.index, newSpan
.length(), perItem) + ", counter:" + mapOutputRecordCounter.getValue());
return newSpan;
}
@@ -815,11 +842,11 @@ public class PipelinedSorter extends ExternalSorter {
return null;
}
int perItem = kvbuffer.position()/items;
- LOG.info(String.format("Span%d.length = %d, perItem = %d", index, length(), perItem));
+ LOG.info(outputContext.getDestinationVertexName() + ": " + String.format("Span%d.length = %d, perItem = %d", index, length(), perItem));
if(remaining.remaining() < METASIZE+perItem) {
//Check if we can get the next Buffer from the main buffer list
if (listIterator.hasNext()) {
- LOG.info("Getting memory from next block in the list, recordsWritten=" +
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "Getting memory from next block in the list, recordsWritten=" +
mapOutputRecordCounter.getValue());
reinit = true;
return listIterator.next();
@@ -1113,10 +1140,10 @@ public class PipelinedSorter extends ExternalSorter {
total += sp.span.length();
eq += sp.span.getEq();
}
- LOG.info("Heap = " + sb.toString());
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "Heap = " + sb.toString());
return true;
} catch(Exception e) {
- LOG.info(e.toString());
+ LOG.info(outputContext.getDestinationVertexName() + ": " + e.toString());
return false;
}
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java
index f58b10b..2a10c35 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java
@@ -129,7 +129,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
final float spillper = this.conf.getFloat(
TezRuntimeConfiguration.TEZ_RUNTIME_SORT_SPILL_PERCENT,
TezRuntimeConfiguration.TEZ_RUNTIME_SORT_SPILL_PERCENT_DEFAULT);
- final int sortmb = computeSortBufferSize((int) availableMemoryMb);
+ final int sortmb = computeSortBufferSize((int) availableMemoryMb, outputContext.getDestinationVertexName());
Preconditions.checkArgument(spillper <= (float) 1.0 && spillper > (float) 0.0,
TezRuntimeConfiguration.TEZ_RUNTIME_SORT_SPILL_PERCENT
@@ -143,7 +143,8 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED_DEFAULT);
if (confPipelinedShuffle) {
- LOG.warn(TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED + " does not work "
+ LOG.warn(outputContext.getDestinationVertexName() + ": " +
+ TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED + " does not work "
+ "with DefaultSorter. It is supported only with PipelinedSorter.");
}
@@ -163,10 +164,14 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
softLimit = (int)(kvbuffer.length * spillper);
bufferRemaining = softLimit;
if (LOG.isInfoEnabled()) {
- LOG.info(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB + ": " + sortmb);
- LOG.info("soft limit at " + softLimit);
- LOG.info("bufstart = " + bufstart + "; bufvoid = " + bufvoid);
- LOG.info("kvstart = " + kvstart + "; length = " + maxRec + "; finalMergeEnabled = " + isFinalMergeEnabled());
+ LOG.info(outputContext.getDestinationVertexName() + ": "
+ + TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB + "=" + sortmb
+ + ", soft limit=" + softLimit
+ + ", bufstart=" + bufstart
+ + ", bufvoid=" + bufvoid
+ + ", kvstart=" + kvstart
+ + ", legnth=" + maxRec
+ + ", finalMergeEnabled=" + isFinalMergeEnabled());
}
// k/v serialization
@@ -176,8 +181,8 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
spillInProgress = false;
minSpillsForCombine = this.conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINE_MIN_SPILLS, 3);
spillThread.setDaemon(true);
- spillThread.setName("SpillThread ["
- + TezUtilsInternal.cleanVertexName(outputContext.getDestinationVertexName() + "]"));
+ spillThread.setName("SpillThread {"
+ + TezUtilsInternal.cleanVertexName(outputContext.getDestinationVertexName() + "}"));
spillLock.lock();
try {
spillThread.start();
@@ -196,7 +201,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
}
@VisibleForTesting
- static int computeSortBufferSize(int availableMemoryMB) {
+ static int computeSortBufferSize(int availableMemoryMB, String logContext) {
if (availableMemoryMB <= 0) {
throw new RuntimeException(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB +
@@ -204,7 +209,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
}
if (availableMemoryMB > MAX_IO_SORT_MB) {
- LOG.warn("Scaling down " + TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB +
+ LOG.warn(logContext + ": Scaling down " + TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB +
"=" + availableMemoryMB + " to " + MAX_IO_SORT_MB
+ " (max sort buffer size supported forDefaultSorter)");
}
@@ -350,7 +355,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
kvindex = (int)(((long)kvindex - NMETA + kvmeta.capacity()) % kvmeta.capacity());
totalKeys++;
} catch (MapBufferTooSmallException e) {
- LOG.info("Record too large for in-memory buffer: " + e.getMessage());
+ LOG.info(outputContext.getDestinationVertexName() + ": Record too large for in-memory buffer: " + e.getMessage());
spillSingleRecord(key, value, partition);
mapOutputRecordCounter.increment(1);
return;
@@ -369,7 +374,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
// Cast one of the operands to long to avoid integer overflow
kvindex = (int) (((long) aligned - METASIZE + kvbuffer.length) % kvbuffer.length) / 4;
if (LOG.isInfoEnabled()) {
- LOG.info("(EQUATOR) " + pos + " kvi " + kvindex +
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "(EQUATOR) " + pos + " kvi " + kvindex +
"(" + (kvindex * 4) + ")");
}
}
@@ -387,7 +392,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
// Cast one of the operands to long to avoid integer overflow
kvstart = kvend = (int) (((long) aligned - METASIZE + kvbuffer.length) % kvbuffer.length) / 4;
if (LOG.isInfoEnabled()) {
- LOG.info("(RESET) equator " + e + " kv " + kvstart + "(" +
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "(RESET) equator " + e + " kv " + kvstart + "(" +
(kvstart * 4) + ")" + " kvi " + kvindex + "(" + (kvindex * 4) + ")");
}
}
@@ -650,13 +655,13 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
kvend = (kvindex + NMETA) % kvmeta.capacity();
bufend = bufmark;
if (LOG.isInfoEnabled()) {
- LOG.info("Sorting & Spilling map output");
- LOG.info("bufstart = " + bufstart + "; bufend = " + bufmark +
- "; bufvoid = " + bufvoid);
- LOG.info("kvstart = " + kvstart + "(" + (kvstart * 4) +
- "); kvend = " + kvend + "(" + (kvend * 4) +
- "); length = " + (distanceTo(kvend, kvstart,
- kvmeta.capacity()) + 1) + "/" + maxRec);
+ LOG.info(
+ outputContext.getDestinationVertexName() + ": " + "Sorting & Spilling map output. "
+ + "bufstart = " + bufstart + ", bufend = " + bufmark + ", bufvoid = " + bufvoid
+ + "; " + "kvstart=" + kvstart + "(" + (kvstart * 4) + ")"
+ + ", kvend = " + kvend + "(" + (kvend * 4) + ")"
+ + ", length = " + (distanceTo(kvend, kvstart, kvmeta.capacity()) + 1) + "/" +
+ maxRec);
}
sortAndSpill();
}
@@ -707,7 +712,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
spillLock.unlock();
sortAndSpill();
} catch (Throwable t) {
- LOG.warn("Got an exception in sortAndSpill", t);
+ LOG.warn(outputContext.getDestinationVertexName() + ": " + "Got an exception in sortAndSpill", t);
sortSpillException = t;
} finally {
spillLock.lock();
@@ -746,13 +751,11 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
bufend = bufmark;
spillInProgress = true;
if (LOG.isInfoEnabled()) {
- LOG.info("Spilling map output");
- LOG.info("bufstart = " + bufstart + "; bufend = " + bufmark +
- "; bufvoid = " + bufvoid);
- LOG.info("kvstart = " + kvstart + "(" + (kvstart * 4) +
- "); kvend = " + kvend + "(" + (kvend * 4) +
- "); length = " + (distanceTo(kvend, kvstart,
- kvmeta.capacity()) + 1) + "/" + maxRec);
+ LOG.info(outputContext.getDestinationVertexName() + ": Spilling map output."
+ + "bufstart=" + bufstart + ", bufend = " + bufmark + ", bufvoid = " + bufvoid
+ +"; kvstart=" + kvstart + "(" + (kvstart * 4) + ")"
+ +", kvend = " + kvend + "(" + (kvend * 4) + ")"
+ + ", length = " + (distanceTo(kvend, kvstart, kvmeta.capacity()) + 1) + "/" + maxRec);
}
spillReady.signal();
}
@@ -849,7 +852,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
TezRawKeyValueIterator kvIter =
new MRResultIterator(spstart, spindex);
if (LOG.isDebugEnabled()) {
- LOG.debug("Running combine processor");
+ LOG.debug(outputContext.getDestinationVertexName() + ": " + "Running combine processor");
}
runCombineProcessor(kvIter, writer);
}
@@ -884,7 +887,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
totalIndexCacheMemory +=
spillRec.size() * MAP_OUTPUT_INDEX_RECORD_LENGTH;
}
- LOG.info("Finished spill " + numSpills);
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "Finished spill " + numSpills);
++numSpills;
if (!isFinalMergeEnabled()) {
numShuffleChunks.setValue(numSpills);
@@ -1063,7 +1066,8 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
ShuffleUtils.generateEventOnSpill(events, isFinalMergeEnabled(), isLastEvent,
outputContext, index, spillRecord, partitions, sendEmptyPartitionDetails, pathComponent);
- LOG.info("Adding spill event for spill (final update=" + isLastEvent + "), spillId=" + index);
+ LOG.info(outputContext.getDestinationVertexName() + ": " +
+ "Adding spill event for spill (final update=" + isLastEvent + "), spillId=" + index);
if (sendEvent) {
outputContext.sendEvents(events);
@@ -1213,7 +1217,8 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
segmentList.add(i, s);
if (LOG.isDebugEnabled()) {
- LOG.debug("TaskIdentifier=" + taskIdentifier + " Partition=" + parts +
+ LOG.debug(outputContext.getDestinationVertexName() + ": "
+ + "TaskIdentifier=" + taskIdentifier + " Partition=" + parts +
"Spill =" + i + "(" + indexRecord.getStartOffset() + "," +
indexRecord.getRawLength() + ", " +
indexRecord.getPartLength() + ")");
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/writers/UnorderedPartitionedKVWriter.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/writers/UnorderedPartitionedKVWriter.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/writers/UnorderedPartitionedKVWriter.java
index 37d8be6..4de167a 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/writers/UnorderedPartitionedKVWriter.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/writers/UnorderedPartitionedKVWriter.java
@@ -91,6 +91,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
// Maybe setup a separate statistics class which can be shared between the
// buffer and the main path instead of having multiple arrays.
+ private final String destNameTrimmed;
private final long availableMemory;
@VisibleForTesting
final WrappedBuffer[] buffers;
@@ -149,6 +150,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
super(outputContext, conf, numOutputs);
Preconditions.checkArgument(availableMemoryBytes >= 0, "availableMemory should be >= 0 bytes");
+ this.destNameTrimmed = TezUtilsInternal.cleanVertexName(outputContext.getDestinationVertexName());
//Not checking for TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT as it might not add much value in
// this case. Add it later if needed.
pipelinedShuffle = this.conf.getBoolean(TezRuntimeConfiguration
@@ -170,13 +172,16 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
int maxSingleBufferSizeBytes = conf.getInt(
TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, Integer.MAX_VALUE);
computeNumBuffersAndSize(maxSingleBufferSizeBytes);
- LOG.info("Running with numBuffers=" + numBuffers + ", sizePerBuffer=" + sizePerBuffer);
+
availableBuffers = new LinkedBlockingQueue<WrappedBuffer>();
buffers = new WrappedBuffer[numBuffers];
// Set up only the first buffer to start with.
buffers[0] = new WrappedBuffer(numOutputs, sizePerBuffer);
numInitializedBuffers = 1;
- LOG.info("Initialize Buffer #" + numInitializedBuffers + " with size=" + sizePerBuffer);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(destNameTrimmed + ": " + "Initializing Buffer #" +
+ numInitializedBuffers + " with size=" + sizePerBuffer);
+ }
currentBuffer = buffers[0];
baos = new ByteArrayOutputStream();
dos = new DataOutputStream(baos);
@@ -189,8 +194,8 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
new ThreadFactoryBuilder()
.setDaemon(true)
.setNameFormat(
- "UnorderedOutSpiller ["
- + TezUtilsInternal.cleanVertexName(outputContext.getDestinationVertexName()) + "]")
+ "UnorderedOutSpiller {"
+ + TezUtilsInternal.cleanVertexName(outputContext.getDestinationVertexName()) + "}")
.build());
spillExecutor = MoreExecutors.listeningDecorator(executor);
numRecordsPerPartition = new int[numPartitions];
@@ -213,7 +218,12 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
skipBuffers = false;
writer = null;
}
- LOG.info("pipelinedShuffle=" + pipelinedShuffle + ", skipBuffers=" + skipBuffers);
+ LOG.info(destNameTrimmed + ": "
+ + "numBuffers=" + numBuffers
+ + ", sizePerBuffer=" + sizePerBuffer
+ + ", skipBuffers=" + skipBuffers
+ + ", pipelinedShuffle=" + pipelinedShuffle
+ + ", numPartitions=" + numPartitions);
}
private void computeNumBuffersAndSize(int bufferLimit) {
@@ -320,7 +330,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
currentBuffer.reset();
} else {
// Update overall stats
- LOG.info("Moving to next buffer and triggering spill");
+ LOG.info(destNameTrimmed + ": " + "Moving to next buffer and triggering spill");
updateGlobalStats(currentBuffer);
pendingSpillCount.incrementAndGet();
@@ -419,10 +429,10 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
spillResult = new SpillResult(compressedLength, this.wrappedBuffer);
handleSpillIndex(spillPathDetails, spillRecord);
- LOG.info("Finished spill " + spillIndex);
+ LOG.info(destNameTrimmed + ": " + "Finished spill " + spillIndex);
if (LOG.isDebugEnabled()) {
- LOG.debug("Spill=" + spillIndex + ", indexPath="
+ LOG.debug(destNameTrimmed + ": " + "Spill=" + spillIndex + ", indexPath="
+ spillPathDetails.indexFilePath + ", outputPath=" + spillPathDetails.outputFilePath);
}
return spillResult;
@@ -460,7 +470,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
isShutdown.set(true);
spillLock.lock();
try {
- LOG.info("Waiting for all spills to complete : Pending : " + pendingSpillCount.get());
+ LOG.info(destNameTrimmed + ": " + "Waiting for all spills to complete : Pending : " + pendingSpillCount.get());
while (pendingSpillCount.get() != 0 && spillException == null) {
spillInProgress.await();
}
@@ -468,7 +478,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
spillLock.unlock();
}
if (spillException != null) {
- LOG.error("Error during spill, throwing");
+ LOG.error(destNameTrimmed + ": " + "Error during spill, throwing");
// Assuming close will be called on the same thread as the write
cleanup();
currentBuffer.cleanup();
@@ -479,7 +489,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
throw new IOException(spillException);
}
} else {
- LOG.info("All spills complete");
+ LOG.info(destNameTrimmed + ": " + "All spills complete");
// Assuming close will be called on the same thread as the write
cleanup();
@@ -686,7 +696,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
for (int i = 0; i < numPartitions; i++) {
long segmentStart = out.getPos();
if (numRecordsPerPartition[i] == 0) {
- LOG.info("Skipping partition: " + i + " in final merge since it has no records");
+ LOG.info(destNameTrimmed + ": " + "Skipping partition: " + i + " in final merge since it has no records");
continue;
}
writer = new Writer(conf, out, keyClass, valClass, codec, null, null);
@@ -738,7 +748,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
}
finalSpillRecord.writeToFile(finalIndexPath, conf);
fileOutputBytesCounter.increment(indexFileSizeEstimate);
- LOG.info("Finished final spill after merging : " + numSpills.get() + " spills");
+ LOG.info(destNameTrimmed + ": " + "Finished final spill after merging : " + numSpills.get() + " spills");
}
private void writeLargeRecord(final Object key, final Object value, final int partition)
@@ -790,9 +800,9 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
sendPipelinedEventForSpill(emptyPartitions, spillIndex, false);
- LOG.info("Finished writing large record of size " + outSize + " to spill file " + spillIndex);
+ LOG.info(destNameTrimmed + ": " + "Finished writing large record of size " + outSize + " to spill file " + spillIndex);
if (LOG.isDebugEnabled()) {
- LOG.debug("LargeRecord Spill=" + spillIndex + ", indexPath="
+ LOG.debug(destNameTrimmed + ": " + "LargeRecord Spill=" + spillIndex + ", indexPath="
+ spillPathDetails.indexFilePath + ", outputPath="
+ spillPathDetails.outputFilePath);
}
@@ -901,10 +911,10 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
Event compEvent = generateDMEvent(true, spillNumber, isFinalUpdate,
pathComponent, emptyPartitions);
- LOG.info("Adding spill event for spill (final update=" + isFinalUpdate + "), spillId=" + spillNumber);
+ LOG.info(destNameTrimmed + ": " + "Adding spill event for spill (final update=" + isFinalUpdate + "), spillId=" + spillNumber);
outputContext.sendEvents(Collections.singletonList(compEvent));
} catch (IOException e) {
- LOG.error("Error in sending pipelined events", e);
+ LOG.error(destNameTrimmed + ": " + "Error in sending pipelined events", e);
outputContext.fatalError(e, "Error in sending pipelined events");
}
}
@@ -924,7 +934,6 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
@Override
public void onSuccess(SpillResult result) {
- LOG.info("Spill# " + spillNumber + " complete.");
spilledSize += result.spillSize;
sendPipelinedEventForSpill(result.wrappedBuffer.recordsPerPartition, spillNumber, false);
@@ -934,7 +943,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
availableBuffers.add(result.wrappedBuffer);
} catch (Throwable e) {
- LOG.error("Failure while attempting to reset buffer after spill", e);
+ LOG.error(destNameTrimmed + ": " + "Failure while attempting to reset buffer after spill", e);
outputContext.fatalError(e, "Failure while attempting to reset buffer after spill");
}
@@ -959,7 +968,7 @@ public class UnorderedPartitionedKVWriter extends BaseUnorderedPartitionedKVWrit
public void onFailure(Throwable t) {
// spillException setup to throw an exception back to the user. Requires synchronization.
// Consider removing it in favor of having Tez kill the task
- LOG.error("Failure while spilling to disk", t);
+ LOG.error(destNameTrimmed + ": " + "Failure while spilling to disk", t);
spillException = t;
outputContext.fatalError(t, "Failure while spilling to disk");
spillLock.lock();
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java
index d784fcd..fa8630d 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/OrderedGroupedKVInput.java
@@ -129,8 +129,10 @@ public class OrderedGroupedKVInput extends AbstractLogicalInput {
List<Event> pending = new LinkedList<Event>();
pendingEvents.drainTo(pending);
if (pending.size() > 0) {
- LOG.info("NoAutoStart delay in processing first event: "
- + (System.currentTimeMillis() - firstEventReceivedTime));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("NoAutoStart delay in processing first event: "
+ + (System.currentTimeMillis() - firstEventReceivedTime));
+ }
shuffle.handleEvents(pending);
}
isStarted.set(true);
@@ -274,10 +276,16 @@ public class OrderedGroupedKVInput extends AbstractLogicalInput {
protected synchronized void createValuesIterator()
throws IOException {
// Not used by ReduceProcessor
- vIter = new ValuesIterator(rawIter,
- (RawComparator) ConfigUtils.getIntermediateInputKeyComparator(conf),
- ConfigUtils.getIntermediateInputKeyClass(conf),
- ConfigUtils.getIntermediateInputValueClass(conf), conf, inputKeyCounter, inputValueCounter);
+ RawComparator rawComparator = ConfigUtils.getIntermediateInputKeyComparator(conf);
+ Class<?> keyClass = ConfigUtils.getIntermediateInputKeyClass(conf);
+ Class<?> valClass = ConfigUtils.getIntermediateInputValueClass(conf);
+ LOG.info(getContext().getSourceVertexName() + ": " + "creating ValuesIterator with "
+ + "comparator=" + rawComparator.getClass().getName()
+ + ", keyClass=" + keyClass.getName()
+ + ", valClass=" + valClass.getName());
+
+ vIter = new ValuesIterator(rawIter, rawComparator, keyClass, valClass,
+ conf, inputKeyCounter, inputValueCounter);
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java
index 62fa9a5..f5c8091 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/input/UnorderedKVInput.java
@@ -24,6 +24,7 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.tez.common.TezUtilsInternal;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.classification.InterfaceAudience;
@@ -131,7 +132,8 @@ public class UnorderedKVInput extends AbstractLogicalInput {
ifileBufferSize = conf.getInt("io.file.buffer.size",
TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_BUFFER_SIZE_DEFAULT);
- this.inputManager = new SimpleFetchedInputAllocator(getContext().getUniqueIdentifier(), conf,
+ this.inputManager = new SimpleFetchedInputAllocator(
+ TezUtilsInternal.cleanVertexName(getContext().getSourceVertexName()), getContext().getUniqueIdentifier(), conf,
getContext().getTotalMemoryAvailableToTask(),
memoryUpdateCallbackHandler.getMemoryAssigned());
@@ -149,8 +151,10 @@ public class UnorderedKVInput extends AbstractLogicalInput {
List<Event> pending = new LinkedList<Event>();
pendingEvents.drainTo(pending);
if (pending.size() > 0) {
- LOG.info("NoAutoStart delay in processing first event: "
- + (System.currentTimeMillis() - firstEventReceivedTime));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() + ": " + "NoAutoStart delay in processing first event: "
+ + (System.currentTimeMillis() - firstEventReceivedTime));
+ }
inputEventHandler.handleEvents(pending);
}
isStarted.set(true);
@@ -207,6 +211,10 @@ public class UnorderedKVInput extends AbstractLogicalInput {
@Override
public synchronized List<Event> close() throws Exception {
+ if (this.inputEventHandler != null) {
+ this.inputEventHandler.logProgress(true);
+ }
+
if (this.shuffleManager != null) {
this.shuffleManager.shutdown();
}
@@ -217,7 +225,7 @@ public class UnorderedKVInput extends AbstractLogicalInput {
long inputRecords = getContext().getCounters()
.findCounter(TaskCounter.INPUT_RECORDS_PROCESSED).getValue();
getContext().getStatisticsReporter().reportItemsProcessed(inputRecords);
-
+
return null;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OrderedPartitionedKVOutput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OrderedPartitionedKVOutput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OrderedPartitionedKVOutput.java
index 6227fb9..2b4c0f4 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OrderedPartitionedKVOutput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/OrderedPartitionedKVOutput.java
@@ -127,7 +127,7 @@ public class OrderedPartitionedKVOutput extends AbstractLogicalOutput {
if (pipelinedShuffle) {
if (finalMergeEnabled) {
- LOG.info("Disabling final merge as "
+ LOG.info(getContext().getDestinationVertexName() + " disabling final merge as "
+ TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED + " is enabled.");
finalMergeEnabled = false;
conf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT, false);
@@ -185,8 +185,8 @@ public class OrderedPartitionedKVOutput extends AbstractLogicalOutput {
this.endTime = System.nanoTime();
returnEvents = generateEvents();
} else {
- LOG.warn(
- "Attempting to close output {} of type {} before it was started. Generating empty events",
+ LOG.warn(getContext().getDestinationVertexName() +
+ ": Attempting to close output {} of type {} before it was started. Generating empty events",
getContext().getDestinationVertexName(), this.getClass().getSimpleName());
returnEvents = generateEmptyEvents();
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedKVOutput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedKVOutput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedKVOutput.java
index 08e6ec0..674f2d5 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedKVOutput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedKVOutput.java
@@ -104,7 +104,7 @@ public class UnorderedKVOutput extends AbstractLogicalOutput {
this.kvWriter = new UnorderedPartitionedKVWriter(getContext(), conf, 1,
memoryUpdateCallbackHandler.getMemoryAssigned());
isStarted.set(true);
- LOG.info(this.getClass().getSimpleName() + " started. MemoryAssigned="
+ LOG.info(getContext().getDestinationVertexName() + " started. MemoryAssigned="
+ memoryUpdateCallbackHandler.getMemoryAssigned());
}
}
@@ -127,8 +127,8 @@ public class UnorderedKVOutput extends AbstractLogicalOutput {
//TODO: Do we need to support sending payloads via events?
returnEvents = kvWriter.close();
} else {
- LOG.warn(
- "Attempting to close output {} of type {} before it was started. Generating empty events",
+ LOG.warn(getContext().getDestinationVertexName() +
+ ": Attempting to close output {} of type {} before it was started. Generating empty events",
getContext().getDestinationVertexName(), this.getClass().getSimpleName());
returnEvents = new LinkedList<Event>();
ShuffleUtils
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedPartitionedKVOutput.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedPartitionedKVOutput.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedPartitionedKVOutput.java
index 38450ee..c60923f 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedPartitionedKVOutput.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/output/UnorderedPartitionedKVOutput.java
@@ -104,8 +104,8 @@ public class UnorderedPartitionedKVOutput extends AbstractLogicalOutput {
if (isStarted.get()) {
returnEvents = kvWriter.close();
} else {
- LOG.warn(
- "Attempting to close output {} of type {} before it was started. Generating empty events",
+ LOG.warn(getContext().getDestinationVertexName() +
+ ": Attempting to close output {} of type {} before it was started. Generating empty events",
getContext().getDestinationVertexName(), this.getClass().getSimpleName());
returnEvents = new LinkedList<Event>();
ShuffleUtils
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestSimpleFetchedInputAllocator.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestSimpleFetchedInputAllocator.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestSimpleFetchedInputAllocator.java
index ffa9429..2f89b0f 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestSimpleFetchedInputAllocator.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/impl/TestSimpleFetchedInputAllocator.java
@@ -52,7 +52,7 @@ public class TestSimpleFetchedInputAllocator {
long inMemThreshold = (long) (bufferPercent * jvmMax);
LOG.info("InMemThreshold: " + inMemThreshold);
- SimpleFetchedInputAllocator inputManager = new SimpleFetchedInputAllocator(UUID.randomUUID().toString(),
+ SimpleFetchedInputAllocator inputManager = new SimpleFetchedInputAllocator("srcName", UUID.randomUUID().toString(),
conf, Runtime.getRuntime().maxMemory(), inMemThreshold);
long requestSize = (long) (0.4f * inMemThreshold);
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestShuffleInputEventHandlerOrderedGrouped.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestShuffleInputEventHandlerOrderedGrouped.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestShuffleInputEventHandlerOrderedGrouped.java
index eed9fd8..6c000ff 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestShuffleInputEventHandlerOrderedGrouped.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/TestShuffleInputEventHandlerOrderedGrouped.java
@@ -155,7 +155,8 @@ public class TestShuffleInputEventHandlerOrderedGrouped {
bytesShuffedToDisk,
bytesShuffedToDiskDirect,
bytesShuffedToMem,
- System.currentTimeMillis());
+ System.currentTimeMillis(),
+ "srcNameTrimmed");
scheduler = spy(realScheduler);
handler = new ShuffleInputEventHandlerOrderedGrouped(inputContext, scheduler, false);
mergeManager = mock(MergeManager.class);
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/sort/impl/dflt/TestDefaultSorter.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/sort/impl/dflt/TestDefaultSorter.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/sort/impl/dflt/TestDefaultSorter.java
index bda2fdd..ecc44a7 100644
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/sort/impl/dflt/TestDefaultSorter.java
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/common/sort/impl/dflt/TestDefaultSorter.java
@@ -213,19 +213,19 @@ public class TestDefaultSorter {
public void testSortMBLimits() throws Exception {
assertTrue("Expected " + DefaultSorter.MAX_IO_SORT_MB,
- DefaultSorter.computeSortBufferSize(4096) == DefaultSorter.MAX_IO_SORT_MB);
+ DefaultSorter.computeSortBufferSize(4096, "") == DefaultSorter.MAX_IO_SORT_MB);
assertTrue("Expected " + DefaultSorter.MAX_IO_SORT_MB,
- DefaultSorter.computeSortBufferSize(2047) == DefaultSorter.MAX_IO_SORT_MB);
- assertTrue("Expected 1024", DefaultSorter.computeSortBufferSize(1024) == 1024);
+ DefaultSorter.computeSortBufferSize(2047, "") == DefaultSorter.MAX_IO_SORT_MB);
+ assertTrue("Expected 1024", DefaultSorter.computeSortBufferSize(1024, "") == 1024);
try {
- DefaultSorter.computeSortBufferSize(0);
+ DefaultSorter.computeSortBufferSize(0, "");
fail("Should have thrown error for setting buffer size to 0");
} catch(RuntimeException re) {
}
try {
- DefaultSorter.computeSortBufferSize(-100);
+ DefaultSorter.computeSortBufferSize(-100, "");
fail("Should have thrown error for setting buffer size to negative value");
} catch(RuntimeException re) {
}
[2/2] tez git commit: TEZ_2844. Backport TEZ-2775 to branch-0.7.
Improve and consolidate logging in Runtime components. (sseth)
Posted by ss...@apache.org.
TEZ_2844. Backport TEZ-2775 to branch-0.7. Improve and consolidate
logging in Runtime components. (sseth)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/cd80d9ab
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/cd80d9ab
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/cd80d9ab
Branch: refs/heads/branch-0.7
Commit: cd80d9ab592f8c116031087da01ac3f6a84a86d2
Parents: c7a946c
Author: Siddharth Seth <ss...@apache.org>
Authored: Tue Sep 22 11:04:25 2015 -0700
Committer: Siddharth Seth <ss...@apache.org>
Committed: Tue Sep 22 11:04:25 2015 -0700
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../resources/tez-container-log4j.properties | 2 +-
.../hadoop/mapred/split/TezGroupedSplit.java | 12 +++
.../hadoop/mapreduce/split/TezGroupedSplit.java | 12 +++
.../org/apache/tez/mapreduce/input/MRInput.java | 36 ++++---
.../tez/mapreduce/input/MRInputLegacy.java | 8 +-
.../tez/mapreduce/input/MultiMRInput.java | 22 ++--
.../apache/tez/mapreduce/output/MROutput.java | 13 ++-
.../runtime/LogicalIOProcessorRuntimeTask.java | 39 ++++++-
.../apache/tez/runtime/task/TaskReporter.java | 2 +-
.../library/common/InputAttemptIdentifier.java | 2 +-
.../runtime/library/common/TezRuntimeUtils.java | 8 +-
.../runtime/library/common/ValuesIterator.java | 2 -
.../runtime/library/common/shuffle/Fetcher.java | 103 +++++++++++++------
.../library/common/shuffle/HttpConnection.java | 4 +-
.../common/shuffle/ShuffleEventHandler.java | 1 +
.../library/common/shuffle/ShuffleUtils.java | 40 +++++--
.../impl/ShuffleInputEventHandlerImpl.java | 41 ++++++--
.../common/shuffle/impl/ShuffleManager.java | 82 +++++++++------
.../impl/SimpleFetchedInputAllocator.java | 26 +++--
.../orderedgrouped/FetcherOrderedGrouped.java | 86 +++++++++++-----
.../shuffle/orderedgrouped/MergeManager.java | 52 +++++-----
.../common/shuffle/orderedgrouped/Shuffle.java | 36 ++++---
.../ShuffleInputEventHandlerOrderedGrouped.java | 42 ++++++--
.../orderedgrouped/ShuffleScheduler.java | 56 ++++++----
.../common/sort/impl/ExternalSorter.java | 26 +++--
.../common/sort/impl/PipelinedSorter.java | 101 +++++++++++-------
.../common/sort/impl/dflt/DefaultSorter.java | 69 +++++++------
.../writers/UnorderedPartitionedKVWriter.java | 49 +++++----
.../library/input/OrderedGroupedKVInput.java | 20 ++--
.../runtime/library/input/UnorderedKVInput.java | 16 ++-
.../output/OrderedPartitionedKVOutput.java | 6 +-
.../library/output/UnorderedKVOutput.java | 6 +-
.../output/UnorderedPartitionedKVOutput.java | 4 +-
.../impl/TestSimpleFetchedInputAllocator.java | 2 +-
...tShuffleInputEventHandlerOrderedGrouped.java | 3 +-
.../sort/impl/dflt/TestDefaultSorter.java | 10 +-
37 files changed, 697 insertions(+), 343 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 2b6204c..9fc3555 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -6,6 +6,7 @@ Release 0.7.1: Unreleased
INCOMPATIBLE CHANGES
ALL CHANGES
+ TEZ-2844. Backport TEZ-2775 to branch-0.7. Improve and consolidate logging in Runtime components.
TEZ-2843. Tez UI: Show error if in progress fails due to AM not reachable
TEZ-2842. Tez UI: Update Tez App details page while in-progress
TEZ-2834. Make Tez preemption resilient to incorrect free resource reported
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-dag/src/main/resources/tez-container-log4j.properties
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/resources/tez-container-log4j.properties b/tez-dag/src/main/resources/tez-container-log4j.properties
index c53994e..4620a78 100644
--- a/tez-dag/src/main/resources/tez-container-log4j.properties
+++ b/tez-dag/src/main/resources/tez-container-log4j.properties
@@ -28,7 +28,7 @@ log4j.appender.CLA=org.apache.tez.common.TezContainerLogAppender
log4j.appender.CLA.containerLogDir=${yarn.app.container.log.dir}
log4j.appender.CLA.layout=org.apache.log4j.PatternLayout
-log4j.appender.CLA.layout.ConversionPattern=%d{ISO8601} [%p] [%t]|| %c{2} %m%n:
+log4j.appender.CLA.layout.ConversionPattern=%d{ISO8601} [%p] [%t] |%c{2}|: %m%n
#
# Event Counter Appender
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplit.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplit.java b/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplit.java
index 4f3a0f2..a9893aa 100644
--- a/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplit.java
+++ b/tez-mapreduce/src/main/java/org/apache/hadoop/mapred/split/TezGroupedSplit.java
@@ -22,6 +22,7 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import org.apache.hadoop.classification.InterfaceAudience.Public;
@@ -175,4 +176,15 @@ public class TezGroupedSplit implements InputSplit, Configurable {
public String getRack() {
return rack;
}
+
+ @Override
+ public String toString() {
+ return "TezGroupedSplit{" +
+ "wrappedSplits=" + wrappedSplits +
+ ", wrappedInputFormatName='" + wrappedInputFormatName + '\'' +
+ ", locations=" + Arrays.toString(locations) +
+ ", rack='" + rack + '\'' +
+ ", length=" + length +
+ '}';
+ }
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplit.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplit.java b/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplit.java
index f85bbcd..430d2ec 100644
--- a/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplit.java
+++ b/tez-mapreduce/src/main/java/org/apache/hadoop/mapreduce/split/TezGroupedSplit.java
@@ -22,6 +22,7 @@ import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import org.apache.hadoop.classification.InterfaceAudience.Public;
@@ -185,4 +186,15 @@ public class TezGroupedSplit extends InputSplit
public String getRack() {
return rack;
}
+
+ @Override
+ public String toString() {
+ return "TezGroupedSplit{" +
+ "wrappedSplits=" + wrappedSplits +
+ ", wrappedInputFormatName='" + wrappedInputFormatName + '\'' +
+ ", locations=" + Arrays.toString(locations) +
+ ", rack='" + rack + '\'' +
+ ", length=" + length +
+ '}';
+ }
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java
index 70365cd..93161cb 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInput.java
@@ -437,8 +437,9 @@ public class MRInput extends MRInputBase {
getContext().inputIsReady();
this.splitInfoViaEvents = jobConf.getBoolean(MRJobConfig.MR_TEZ_SPLITS_VIA_EVENTS,
MRJobConfig.MR_TEZ_SPLITS_VIA_EVENTS_DEFAULT);
- LOG.info("Using New mapreduce API: " + useNewApi
- + ", split information via event: " + splitInfoViaEvents);
+ LOG.info(getContext().getSourceVertexName() + " using newmapreduce API=" + useNewApi +
+ ", split via event=" + splitInfoViaEvents + ", numPhysicalInputs=" +
+ getNumPhysicalInputs());
initializeInternal();
return null;
}
@@ -447,7 +448,6 @@ public class MRInput extends MRInputBase {
public void start() {
Preconditions.checkState(getNumPhysicalInputs() == 0 || getNumPhysicalInputs() == 1,
"Expecting 0 or 1 physical input for MRInput");
- LOG.info("MRInput setup to received {} events", getNumPhysicalInputs());
}
@Private
@@ -490,7 +490,7 @@ public class MRInput extends MRInputBase {
} finally {
rrLock.unlock();
}
- LOG.info("Initialzed MRInput: " + getContext().getSourceVertexName());
+ LOG.info("Initialized MRInput: " + getContext().getSourceVertexName());
}
/**
@@ -588,7 +588,9 @@ public class MRInput extends MRInputBase {
rrLock.lock();
try {
initFromEventInternal(event);
- LOG.info("Notifying on RecordReader Initialized");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() + " notifying on RecordReader initialized");
+ }
rrInited.signal();
} finally {
rrLock.unlock();
@@ -599,7 +601,9 @@ public class MRInput extends MRInputBase {
assert rrLock.getHoldCount() == 1;
rrLock.lock();
try {
- LOG.info("Awaiting RecordReader initialization");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() + " awaiting RecordReader initialization");
+ }
rrInited.await();
} catch (Exception e) {
throw new IOException(
@@ -621,22 +625,30 @@ public class MRInput extends MRInputBase {
}
private void initFromEventInternal(InputDataInformationEvent initEvent) throws IOException {
- LOG.info("Initializing RecordReader from event");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() + " initializing RecordReader from event");
+ }
Preconditions.checkState(initEvent != null, "InitEvent must be specified");
MRSplitProto splitProto = MRSplitProto.parseFrom(ByteString.copyFrom(initEvent.getUserPayload()));
Object split = null;
if (useNewApi) {
split = MRInputUtils.getNewSplitDetailsFromEvent(splitProto, jobConf);
- LOG.info("Split Details -> SplitClass: " + split.getClass().getName() + ", NewSplit: "
- + split);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() + " split Details -> SplitClass: " +
+ split.getClass().getName() + ", NewSplit: "
+ + split);
+ }
} else {
split = MRInputUtils.getOldSplitDetailsFromEvent(splitProto, jobConf);
- LOG.info("Split Details -> SplitClass: " + split.getClass().getName() + ", OldSplit: "
- + split);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() + " split Details -> SplitClass: " +
+ split.getClass().getName() + ", OldSplit: "
+ + split);
+ }
}
mrReader.setSplit(split);
- LOG.info("Initialized RecordReader from event");
+ LOG.info(getContext().getSourceVertexName() + " initialized RecordReader from event");
}
private static class MRInputHelpersInternal extends MRInputHelpers {
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInputLegacy.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInputLegacy.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInputLegacy.java
index d825d53..e83c36a 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInputLegacy.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MRInputLegacy.java
@@ -73,7 +73,7 @@ public class MRInputLegacy extends MRInput {
@Private
protected void initializeInternal() throws IOException {
- LOG.info("MRInputLegacy deferring initialization");
+ LOG.info(getContext().getSourceVertexName() + " MRInputLegacy deferring initialization");
}
@Private
@@ -130,7 +130,11 @@ public class MRInputLegacy extends MRInput {
}
if (splitInfoViaEvents && !inited) {
if (initEvent == null) {
- LOG.info("Awaiting init event before initializing record reader");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() +
+ " awaiting init event before initializing record reader");
+ }
+
try {
eventCondition.await();
} catch (InterruptedException e) {
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MultiMRInput.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MultiMRInput.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MultiMRInput.java
index 44d9c96..4a792dc 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MultiMRInput.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/MultiMRInput.java
@@ -108,8 +108,8 @@ public class MultiMRInput extends MRInputBase {
@Override
public List<Event> initialize() throws IOException {
super.initialize();
- LOG.info("Using New mapreduce API: " + useNewApi + ", numPhysicalInputs: "
- + getNumPhysicalInputs());
+ LOG.info(getContext().getSourceVertexName() + " using newmapreduce API=" + useNewApi +
+ ", numPhysicalInputs=" + getNumPhysicalInputs());
if (getNumPhysicalInputs() == 0) {
getContext().inputIsReady();
}
@@ -164,7 +164,9 @@ public class MultiMRInput extends MRInputBase {
private MRReader initFromEvent(InputDataInformationEvent event) throws IOException {
Preconditions.checkState(event != null, "Event must be specified");
- LOG.info("Initializing Reader: " + eventCount.get());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() + " initializing Reader: " + eventCount.get());
+ }
MRSplitProto splitProto = MRSplitProto.parseFrom(ByteString.copyFrom(event.getUserPayload()));
Object split = null;
MRReader reader = null;
@@ -176,17 +178,21 @@ public class MultiMRInput extends MRInputBase {
.getClusterTimestamp(), getContext().getTaskVertexIndex(), getContext()
.getApplicationId().getId(), getContext().getTaskIndex(), getContext()
.getTaskAttemptNumber());
- LOG.info("Split Details -> SplitClass: " + split.getClass().getName() + ", NewSplit: "
- + split);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() + " split Details -> SplitClass: " +
+ split.getClass().getName() + ", NewSplit: " + split);
+ }
} else {
split = MRInputUtils.getOldSplitDetailsFromEvent(splitProto, localJobConf);
reader = new MRReaderMapred(localJobConf, (org.apache.hadoop.mapred.InputSplit) split,
getContext().getCounters(), inputRecordCounter);
- LOG.info("Split Details -> SplitClass: " + split.getClass().getName() + ", OldSplit: "
- + split);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getContext().getSourceVertexName() + " split Details -> SplitClass: " +
+ split.getClass().getName() + ", OldSplit: " + split);
+ }
}
- LOG.info("Initialized RecordReader from event");
+ LOG.info(getContext().getSourceVertexName() + " initialized RecordReader from event");
return reader;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
index d19f707..7136482 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
@@ -343,7 +343,6 @@ public class MROutput extends AbstractLogicalOutput {
@Override
public List<Event> initialize() throws IOException, InterruptedException {
- LOG.info("Initializing Simple Output");
getContext().requestInitialMemory(0l, null); //mandatory call
taskNumberFormat.setMinimumIntegerDigits(5);
taskNumberFormat.setGroupingUsed(false);
@@ -380,6 +379,8 @@ public class MROutput extends AbstractLogicalOutput {
}
}
+ String outputFormatClassName;
+
outputRecordCounter = getContext().getCounters().findCounter(TaskCounter.OUTPUT_RECORDS);
if (useNewApi) {
@@ -388,6 +389,7 @@ public class MROutput extends AbstractLogicalOutput {
newOutputFormat =
org.apache.hadoop.util.ReflectionUtils.newInstance(
newApiTaskAttemptContext.getOutputFormatClass(), jobConf);
+ outputFormatClassName = newOutputFormat.getClass().getName();
} catch (ClassNotFoundException cnfe) {
throw new IOException(cnfe);
}
@@ -404,6 +406,7 @@ public class MROutput extends AbstractLogicalOutput {
jobConf, taskAttemptId,
new MRTaskReporter(getContext()));
oldOutputFormat = jobConf.getOutputFormat();
+ outputFormatClassName = oldOutputFormat.getClass().getName();
FileSystem fs = FileSystem.get(jobConf);
String finalName = getOutputName();
@@ -414,8 +417,9 @@ public class MROutput extends AbstractLogicalOutput {
}
initCommitter(jobConf, useNewApi);
- LOG.info("Initialized Simple Output"
- + ", using_new_api: " + useNewApi);
+ LOG.info(getContext().getDestinationVertexName() + ": "
+ + "outputFormat=" + outputFormatClassName
+ + ", using newmapreduce API=" + useNewApi);
return null;
}
@@ -517,6 +521,7 @@ public class MROutput extends AbstractLogicalOutput {
@Override
public synchronized List<Event> close() throws IOException {
flush();
+ LOG.info(getContext().getDestinationVertexName() + " closed");
long outputRecords = getContext().getCounters()
.findCounter(TaskCounter.OUTPUT_RECORDS).getValue();
getContext().getStatisticsReporter().reportItemsProcessed(outputRecords);
@@ -534,7 +539,6 @@ public class MROutput extends AbstractLogicalOutput {
return;
}
- LOG.info("Flushing Simple Output");
if (useNewApi) {
try {
newRecordWriter.close(newApiTaskAttemptContext);
@@ -544,7 +548,6 @@ public class MROutput extends AbstractLogicalOutput {
} else {
oldRecordWriter.close(null);
}
- LOG.info("Flushed Simple Output");
}
/**
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java
index 92035e1..c61bb4e 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java
@@ -404,6 +404,17 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask {
@Override
protected Void callInternal() throws Exception {
+ String oldThreadName = Thread.currentThread().getName();
+ try {
+ Thread.currentThread().setName(oldThreadName + "{" + inputSpec.getSourceVertexName() + "}");
+ return _callInternal();
+ } finally {
+ Thread.currentThread().setName(oldThreadName);
+ }
+ }
+
+ protected Void _callInternal() throws Exception {
+
if (LOG.isDebugEnabled()) {
LOG.debug("Initializing Input using InputSpec: " + inputSpec);
}
@@ -437,6 +448,17 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask {
@Override
protected Void callInternal() throws Exception {
+ String oldThreadName = Thread.currentThread().getName();
+ try {
+ Thread.currentThread().setName(oldThreadName + " Start: {" + srcVertexName + "}");
+ return _callInternal();
+ } finally {
+ Thread.currentThread().setName(oldThreadName);
+ }
+ }
+
+ protected Void _callInternal() throws Exception {
+ Thread.currentThread().setName("InitializerStart {" + srcVertexName + "}");
if (LOG.isDebugEnabled()) {
LOG.debug("Starting Input with src edge: " + srcVertexName);
}
@@ -459,6 +481,17 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask {
@Override
protected Void callInternal() throws Exception {
+ String oldThreadName = Thread.currentThread().getName();
+ try {
+ Thread.currentThread().setName(oldThreadName + "{" + outputSpec.getDestinationVertexName() + "}");
+ return _callInternal();
+ } finally {
+ Thread.currentThread().setName(oldThreadName);
+ }
+ }
+
+ protected Void _callInternal() throws Exception {
+ Thread.currentThread().setName("Initializer {" + outputSpec.getDestinationVertexName() + "}");
if (LOG.isDebugEnabled()) {
LOG.debug("Initializing Output using OutputSpec: " + outputSpec);
}
@@ -727,8 +760,8 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask {
}
});
- eventRouterThread.setName("TezTaskEventRouter["
- + taskSpec.getTaskAttemptID().toString() + "]");
+ eventRouterThread.setName("TezTaskEventRouter{"
+ + taskSpec.getTaskAttemptID().toString() + "}");
eventRouterThread.start();
}
@@ -756,7 +789,7 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask {
}
public void cleanup() throws InterruptedException {
- LOG.info("Final Counters : " + getCounters().toShortString());
+ LOG.info("Final Counters for " + taskSpec.getTaskAttemptID() + ": " + getCounters().toShortString());
setTaskDone();
if (eventRouterThread != null) {
eventRouterThread.interrupt();
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java
----------------------------------------------------------------------
diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java
index d9a7786..bf93ce3 100644
--- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java
+++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java
@@ -278,7 +278,7 @@ public class TaskReporter {
int numEventsReceived = 0;
if (task.isTaskDone() || task.hadFatalError()) {
if (response.getEvents() != null && !response.getEvents().isEmpty()) {
- LOG.info("Current task already complete, Ignoring all event in"
+ LOG.info("Current task already complete, Ignoring all events in"
+ " heartbeat response, eventCount=" + response.getEvents().size());
}
} else {
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/InputAttemptIdentifier.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/InputAttemptIdentifier.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/InputAttemptIdentifier.java
index 59fb638..d70942c 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/InputAttemptIdentifier.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/InputAttemptIdentifier.java
@@ -146,6 +146,6 @@ public class InputAttemptIdentifier {
public String toString() {
return "InputAttemptIdentifier [inputIdentifier=" + inputIdentifier
+ ", attemptNumber=" + attemptNumber + ", pathComponent="
- + pathComponent + ", fetchTypeInfo=" + fetchTypeInfo + ", spillEventId=" + spillEventId +"]";
+ + pathComponent + ", spillType=" + fetchTypeInfo.ordinal() + ", spillId=" + spillEventId +"]";
}
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/TezRuntimeUtils.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/TezRuntimeUtils.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/TezRuntimeUtils.java
index 69436ba..819423f 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/TezRuntimeUtils.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/TezRuntimeUtils.java
@@ -64,7 +64,9 @@ public class TezRuntimeUtils {
if (className == null) {
return null;
}
- LOG.info("Using Combiner class: " + className);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Using Combiner class: " + className);
+ }
try {
clazz = (Class<? extends Combiner>) conf.getClassByName(className);
} catch (ClassNotFoundException e) {
@@ -105,7 +107,9 @@ public class TezRuntimeUtils {
+ conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS), e);
}
- LOG.info("Using partitioner class: " + clazz.getName());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Using partitioner class: " + clazz.getName());
+ }
Partitioner partitioner = null;
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ValuesIterator.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ValuesIterator.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ValuesIterator.java
index 24f9f8a..f4da742 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ValuesIterator.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/ValuesIterator.java
@@ -45,7 +45,6 @@ import com.google.common.base.Preconditions;
@Private
public class ValuesIterator<KEY,VALUE> {
- private static final Logger LOG = LoggerFactory.getLogger(ValuesIterator.class.getName());
protected TezRawKeyValueIterator in; //input iterator
private KEY key; // current key
private KEY nextKey;
@@ -82,7 +81,6 @@ public class ValuesIterator<KEY,VALUE> {
this.keyDeserializer.open(keyIn);
this.valDeserializer = serializationFactory.getDeserializer(valClass);
this.valDeserializer.open(this.valueIn);
- LOG.info("keyDeserializer=" + keyDeserializer + "; valueDeserializer=" + valDeserializer);
}
TezRawKeyValueIterator getRawIterator() { return in; }
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
index 0dee90c..be68cc1 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/Fetcher.java
@@ -202,8 +202,10 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
fetcherCallback.fetchFailed(host, left, hostFetchResult.connectFailed);
}
} else {
- LOG.info("Ignoring failed fetch reports for " + hostFetchResult.failedInputs.length +
- " inputs since the fetcher has already been stopped");
+ if (isDebugEnabled) {
+ LOG.debug("Ignoring failed fetch reports for " + hostFetchResult.failedInputs.length +
+ " inputs since the fetcher has already been stopped");
+ }
}
}
@@ -409,9 +411,11 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
// indirectly penalizing the host
InputAttemptIdentifier[] failedFetches = null;
if (isShutDown.get()) {
- LOG.info(
- "Not reporting fetch failure during connection establishment, since an Exception was caught after shutdown." +
- e.getClass().getName() + ", Message: " + e.getMessage());
+ if (isDebugEnabled) {
+ LOG.debug(
+ "Not reporting fetch failure during connection establishment, since an Exception was caught after shutdown." +
+ e.getClass().getName() + ", Message: " + e.getMessage());
+ }
} else {
failedFetches = remaining.toArray(new InputAttemptIdentifier[remaining.size()]);
}
@@ -420,7 +424,9 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
if (isShutDown.get()) {
// shutdown would have no effect if in the process of establishing the connection.
shutdownInternal();
- LOG.info("Detected fetcher has been shutdown after connection establishment. Returning");
+ if (isDebugEnabled) {
+ LOG.debug("Detected fetcher has been shutdown after connection establishment. Returning");
+ }
return new HostFetchResult(new FetchResult(host, port, partition, remaining), null, false);
}
@@ -434,9 +440,11 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
// with the first map, typically lost map. So, penalize only that map
// and add the rest
if (isShutDown.get()) {
- LOG.info(
- "Not reporting fetch failure during connection establishment, since an Exception was caught after shutdown." +
- e.getClass().getName() + ", Message: " + e.getMessage());
+ if (isDebugEnabled) {
+ LOG.debug(
+ "Not reporting fetch failure during connection establishment, since an Exception was caught after shutdown." +
+ e.getClass().getName() + ", Message: " + e.getMessage());
+ }
} else {
InputAttemptIdentifier firstAttempt = attempts.get(0);
LOG.warn("Fetch Failure from host while connecting: " + host + ", attempt: " + firstAttempt
@@ -462,7 +470,9 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
if (isShutDown.get()) {
// shutdown would have no effect if in the process of establishing the connection.
shutdownInternal();
- LOG.info("Detected fetcher has been shutdown after opening stream. Returning");
+ if (isDebugEnabled) {
+ LOG.debug("Detected fetcher has been shutdown after opening stream. Returning");
+ }
return new HostFetchResult(new FetchResult(host, port, partition, remaining), null, false);
}
// After this point, closing the stream and connection, should cause a
@@ -477,7 +487,9 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
while (!remaining.isEmpty() && failedInputs == null) {
if (isShutDown.get()) {
shutdownInternal(true);
- LOG.info("Fetcher already shutdown. Aborting queued fetches for " + remaining.size() + " inputs");
+ if (isDebugEnabled) {
+ LOG.debug("Fetcher already shutdown. Aborting queued fetches for " + remaining.size() + " inputs");
+ }
return new HostFetchResult(new FetchResult(host, port, partition, remaining), null,
false);
}
@@ -487,7 +499,9 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
//clean up connection
shutdownInternal(true);
if (isShutDown.get()) {
- LOG.info("Fetcher already shutdown. Aborting reconnection and queued fetches for " + remaining.size() + " inputs");
+ if (isDebugEnabled) {
+ LOG.debug("Fetcher already shutdown. Aborting reconnection and queued fetches for " + remaining.size() + " inputs");
+ }
return new HostFetchResult(new FetchResult(host, port, partition, remaining), null,
false);
}
@@ -501,8 +515,10 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
}
if (isShutDown.get() && failedInputs != null && failedInputs.length > 0) {
- LOG.info("Fetcher already shutdown. Not reporting fetch failures for: " +
- failedInputs.length + " failed inputs");
+ if (isDebugEnabled) {
+ LOG.debug("Fetcher already shutdown. Not reporting fetch failures for: " +
+ failedInputs.length + " failed inputs");
+ }
failedInputs = null;
}
return new HostFetchResult(new FetchResult(host, port, partition, remaining), failedInputs,
@@ -520,7 +536,9 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
Iterator<InputAttemptIdentifier> iterator = remaining.iterator();
while (iterator.hasNext()) {
if (isShutDown.get()) {
- LOG.info("Already shutdown. Skipping fetch for " + remaining.size() + " inputs");
+ if (isDebugEnabled) {
+ LOG.debug("Already shutdown. Skipping fetch for " + remaining.size() + " inputs");
+ }
break;
}
InputAttemptIdentifier srcAttemptId = iterator.next();
@@ -545,9 +563,11 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
@Override
public void freeResources(FetchedInput fetchedInput) {}
});
- LOG.info("fetcher" + " about to shuffle output of srcAttempt (direct disk)" + srcAttemptId
- + " decomp: " + idxRecord.getRawLength() + " len: " + idxRecord.getPartLength()
- + " to " + fetchedInput.getType());
+ if (isDebugEnabled) {
+ LOG.debug("fetcher" + " about to shuffle output of srcAttempt (direct disk)" + srcAttemptId
+ + " decomp: " + idxRecord.getRawLength() + " len: " + idxRecord.getPartLength()
+ + " to " + fetchedInput.getType());
+ }
long endTime = System.currentTimeMillis();
fetcherCallback.fetchSucceeded(host, srcAttemptId, fetchedInput, idxRecord.getPartLength(),
@@ -556,9 +576,12 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
} catch (IOException e) {
cleanupFetchedInput(fetchedInput);
if (isShutDown.get()) {
- LOG.info(
- "Already shutdown. Ignoring Local Fetch Failure for " + srcAttemptId + " from host " +
- host + " : " + e.getClass().getName() + ", message=" + e.getMessage());
+ if (isDebugEnabled) {
+ LOG.debug(
+ "Already shutdown. Ignoring Local Fetch Failure for " + srcAttemptId +
+ " from host " +
+ host + " : " + e.getClass().getName() + ", message=" + e.getMessage());
+ }
break;
}
LOG.warn("Failed to shuffle output of " + srcAttemptId + " from " + host + "(local fetch)",
@@ -569,8 +592,10 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
InputAttemptIdentifier[] failedFetches = null;
if (failMissing && remaining.size() > 0) {
if (isShutDown.get()) {
- LOG.info("Already shutdown, not reporting fetch failures for: " + remaining.size() +
+ if (isDebugEnabled) {
+ LOG.debug("Already shutdown, not reporting fetch failures for: " + remaining.size() +
" remaining inputs");
+ }
} else {
failedFetches = remaining.toArray(new InputAttemptIdentifier[remaining.size()]);
}
@@ -623,6 +648,9 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
public void shutdown() {
if (!isShutDown.getAndSet(true)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Shutting down fetcher for host: " + host);
+ }
shutdownInternal();
}
}
@@ -643,7 +671,7 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
} catch (IOException e) {
LOG.info("Exception while shutting down fetcher on " + logIdentifier + " : "
+ e.getMessage());
- if (LOG.isDebugEnabled()) {
+ if (isDebugEnabled) {
LOG.debug(StringUtils.EMPTY, e);
}
}
@@ -678,7 +706,9 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
// Don't know which one was bad, so consider all of them as bad
return remaining.toArray(new InputAttemptIdentifier[remaining.size()]);
} else {
- LOG.info("Already shutdown. Ignoring badId error with message: " + e.getMessage());
+ if (isDebugEnabled) {
+ LOG.debug("Already shutdown. Ignoring badId error with message: " + e.getMessage());
+ }
return null;
}
}
@@ -694,12 +724,14 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
assert (srcAttemptId != null);
return new InputAttemptIdentifier[]{srcAttemptId};
} else {
- LOG.info("Already shutdown. Ignoring verification failure.");
+ if (isDebugEnabled) {
+ LOG.debug("Already shutdown. Ignoring verification failure.");
+ }
return null;
}
}
- if (LOG.isDebugEnabled()) {
+ if (isDebugEnabled) {
LOG.debug("header: " + srcAttemptId + ", len: " + compressedLength
+ ", decomp len: " + decompressedLength);
}
@@ -723,10 +755,12 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
// }
// Go!
- LOG.info("fetcher" + " about to shuffle output of srcAttempt "
- + fetchedInput.getInputAttemptIdentifier() + " decomp: "
- + decompressedLength + " len: " + compressedLength + " to "
- + fetchedInput.getType());
+ if (isDebugEnabled) {
+ LOG.debug("fetcher" + " about to shuffle output of srcAttempt "
+ + fetchedInput.getInputAttemptIdentifier() + " decomp: "
+ + decompressedLength + " len: " + compressedLength + " to "
+ + fetchedInput.getType());
+ }
if (fetchedInput.getType() == Type.MEMORY) {
ShuffleUtils.shuffleToMemory(((MemoryFetchedInput) fetchedInput).getBytes(),
@@ -735,7 +769,7 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
fetchedInput.getInputAttemptIdentifier().toString());
} else if (fetchedInput.getType() == Type.DISK) {
ShuffleUtils.shuffleToDisk(((DiskFetchedInput) fetchedInput).getOutputStream(),
- (host +":" +port), input, compressedLength, LOG,
+ (host +":" +port), input, compressedLength, decompressedLength, LOG,
fetchedInput.getInputAttemptIdentifier().toString());
} else {
throw new TezUncheckedException("Bad fetchedInput type while fetching shuffle data " +
@@ -765,8 +799,11 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
} catch (IOException ioe) {
if (isShutDown.get()) {
cleanupFetchedInput(fetchedInput);
- LOG.info("Already shutdown. Ignoring exception during fetch " + ioe.getClass().getName() +
- ", Message: " + ioe.getMessage());
+ if (isDebugEnabled) {
+ LOG.debug(
+ "Already shutdown. Ignoring exception during fetch " + ioe.getClass().getName() +
+ ", Message: " + ioe.getMessage());
+ }
return null;
}
if (shouldRetry(srcAttemptId, ioe)) {
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/HttpConnection.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/HttpConnection.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/HttpConnection.java
index ad6ed19..17e6e90 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/HttpConnection.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/HttpConnection.java
@@ -274,7 +274,9 @@ public class HttpConnection {
stopWatch.reset().start();
try {
if (input != null) {
- LOG.info("Closing input on " + logIdentifier);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Closing input on " + logIdentifier);
+ }
input.close();
}
if (httpConnParams.keepAlive && connectionSucceeed) {
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleEventHandler.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleEventHandler.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleEventHandler.java
index ff66158..da7c944 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleEventHandler.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleEventHandler.java
@@ -25,4 +25,5 @@ import org.apache.tez.runtime.api.Event;
public interface ShuffleEventHandler {
public void handleEvents(List<Event> events) throws IOException;
+ public void logProgress(boolean updateOnClose);
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleUtils.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleUtils.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleUtils.java
index 46489ed..1873485 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleUtils.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/ShuffleUtils.java
@@ -107,12 +107,16 @@ public class ShuffleUtils {
Logger LOG, String identifier) throws IOException {
try {
IFile.Reader.readToMemory(shuffleData, input, compressedLength, codec,
- ifileReadAhead, ifileReadAheadLength);
+ ifileReadAhead, ifileReadAheadLength);
// metrics.inputBytes(shuffleData.length);
- LOG.info("Read " + shuffleData.length + " bytes from input for "
- + identifier);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Read " + shuffleData.length + " bytes from input for "
+ + identifier);
+ }
} catch (IOException ioe) {
// Close the streams
+ LOG.info("Failed to read data to memory for " + identifier + ". len=" + compressedLength +
+ ", decomp=" + decompressedLength + ". ExceptionMessage=" + ioe.getMessage());
ioCleanup(input);
// Re-throw
throw ioe;
@@ -120,7 +124,7 @@ public class ShuffleUtils {
}
public static void shuffleToDisk(OutputStream output, String hostIdentifier,
- InputStream input, long compressedLength, Logger LOG, String identifier)
+ InputStream input, long compressedLength, long decompressedLength, Logger LOG, String identifier)
throws IOException {
// Copy data to local-disk
long bytesLeft = compressedLength;
@@ -138,12 +142,16 @@ public class ShuffleUtils {
// metrics.inputBytes(n);
}
- LOG.info("Read " + (compressedLength - bytesLeft)
- + " bytes from input for " + identifier);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Read " + (compressedLength - bytesLeft)
+ + " bytes from input for " + identifier);
+ }
output.close();
} catch (IOException ioe) {
// Close the streams
+ LOG.info("Failed to read data to disk for " + identifier + ". len=" + compressedLength +
+ ", decomp=" + decompressedLength + ". ExceptionMessage=" + ioe.getMessage());
ioCleanup(input, output);
// Re-throw
throw ioe;
@@ -468,10 +476,26 @@ public class ShuffleUtils {
}
log.info(
"Completed fetch for attempt: "
- + srcAttemptIdentifier + " to " + outputType +
- ", CompressedSize=" + bytesCompressed + ", DecompressedSize=" + bytesDecompressed +
+ + toShortString(srcAttemptIdentifier)
+ +" to " + outputType +
+ ", csize=" + bytesCompressed + ", dsize=" + bytesDecompressed +
", EndTime=" + System.currentTimeMillis() + ", TimeTaken=" + millis + ", Rate=" +
MBPS_FORMAT.get().format(rate) + " MB/s");
}
+
+ private static String toShortString(InputAttemptIdentifier inputAttemptIdentifier) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("{");
+ sb.append(inputAttemptIdentifier.getInputIdentifier().getInputIndex());
+ sb.append(", ").append(inputAttemptIdentifier.getAttemptNumber());
+ sb.append(", ").append(inputAttemptIdentifier.getPathComponent());
+ if (inputAttemptIdentifier.getFetchTypeInfo()
+ != InputAttemptIdentifier.SPILL_INFO.FINAL_MERGE_ENABLED) {
+ sb.append(", ").append(inputAttemptIdentifier.getFetchTypeInfo().ordinal());
+ sb.append(", ").append(inputAttemptIdentifier.getSpillEventId());
+ }
+ sb.append("}");
+ return sb.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleInputEventHandlerImpl.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleInputEventHandlerImpl.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleInputEventHandlerImpl.java
index 61b3e3a..8fb1568 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleInputEventHandlerImpl.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleInputEventHandlerImpl.java
@@ -22,6 +22,7 @@ package org.apache.tez.runtime.library.common.shuffle.impl;
import java.io.IOException;
import java.util.BitSet;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
import com.google.protobuf.ByteString;
@@ -32,20 +33,15 @@ import org.apache.tez.common.TezCommonUtils;
import org.apache.tez.common.TezUtilsInternal;
import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.runtime.api.Event;
-import org.apache.tez.runtime.api.Input;
import org.apache.tez.runtime.api.InputContext;
import org.apache.tez.runtime.api.events.DataMovementEvent;
import org.apache.tez.runtime.api.events.InputFailedEvent;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
import org.apache.tez.runtime.library.common.InputIdentifier;
-import org.apache.tez.runtime.library.common.shuffle.DiskFetchedInput;
-import org.apache.tez.runtime.library.common.shuffle.FetchedInput;
import org.apache.tez.runtime.library.common.shuffle.FetchedInputAllocator;
-import org.apache.tez.runtime.library.common.shuffle.MemoryFetchedInput;
import org.apache.tez.runtime.library.common.shuffle.ShuffleEventHandler;
import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
import org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads.DataMovementEventPayloadProto;
-import org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads.DataProto;
import com.google.protobuf.InvalidProtocolBufferException;
@@ -57,16 +53,24 @@ public class ShuffleInputEventHandlerImpl implements ShuffleEventHandler {
private static final Logger LOG = LoggerFactory.getLogger(ShuffleInputEventHandlerImpl.class);
private final ShuffleManager shuffleManager;
+ //TODO: unused. Consider removing later?
private final FetchedInputAllocator inputAllocator;
private final CompressionCodec codec;
private final boolean ifileReadAhead;
private final int ifileReadAheadLength;
private final boolean useSharedInputs;
+ private final InputContext inputContext;
+
+ private final AtomicInteger nextToLogEventCount = new AtomicInteger(0);
+ private final AtomicInteger numDmeEvents = new AtomicInteger(0);
+ private final AtomicInteger numObsoletionEvents = new AtomicInteger(0);
+ private final AtomicInteger numDmeEventsNoData = new AtomicInteger(0);
public ShuffleInputEventHandlerImpl(InputContext inputContext,
ShuffleManager shuffleManager,
FetchedInputAllocator inputAllocator, CompressionCodec codec,
boolean ifileReadAhead, int ifileReadAheadLength) {
+ this.inputContext = inputContext;
this.shuffleManager = shuffleManager;
this.inputAllocator = inputAllocator;
this.codec = codec;
@@ -86,13 +90,29 @@ public class ShuffleInputEventHandlerImpl implements ShuffleEventHandler {
private void handleEvent(Event event) throws IOException {
if (event instanceof DataMovementEvent) {
+ numDmeEvents.incrementAndGet();
processDataMovementEvent((DataMovementEvent)event);
shuffleManager.updateEventReceivedTime();
} else if (event instanceof InputFailedEvent) {
- processInputFailedEvent((InputFailedEvent)event);
+ numObsoletionEvents.incrementAndGet();
+ processInputFailedEvent((InputFailedEvent) event);
} else {
throw new TezUncheckedException("Unexpected event type: " + event.getClass().getName());
}
+ if (numDmeEvents.get() + numObsoletionEvents.get() > nextToLogEventCount.get()) {
+ logProgress(false);
+ // Log every 50 events seen.
+ nextToLogEventCount.addAndGet(50);
+ }
+ }
+
+ @Override
+ public void logProgress(boolean updateOnClose) {
+ LOG.info(inputContext.getSourceVertexName() + ": "
+ + "numDmeEventsSeen=" + numDmeEvents.get()
+ + ", numDmeEventsSeenWithNoData=" + numDmeEventsNoData.get()
+ + ", numObsoletionEventsSeen=" + numObsoletionEvents.get()
+ + (updateOnClose == true ? ", updateOnClose" : ""));
}
private void processDataMovementEvent(DataMovementEvent dme) throws IOException {
@@ -104,9 +124,11 @@ public class ShuffleInputEventHandlerImpl implements ShuffleEventHandler {
throw new TezUncheckedException("Unable to parse DataMovementEvent payload", e);
}
int srcIndex = dme.getSourceIndex();
- LOG.info("DME srcIdx: " + srcIndex + ", targetIndex: " + dme.getTargetIndex()
- + ", attemptNum: " + dme.getVersion() + ", payload: " + ShuffleUtils
- .stringify(shufflePayload));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("DME srcIdx: " + srcIndex + ", targetIndex: " + dme.getTargetIndex()
+ + ", attemptNum: " + dme.getVersion() + ", payload: " + ShuffleUtils
+ .stringify(shufflePayload));
+ }
if (shufflePayload.hasEmptyPartitions()) {
byte[] emptyPartitions = TezCommonUtils.decompressByteStringToByteArray(shufflePayload
@@ -119,6 +141,7 @@ public class ShuffleInputEventHandlerImpl implements ShuffleEventHandler {
LOG.debug("Source partition: " + srcIndex + " did not generate any data. SrcAttempt: ["
+ srcAttemptIdentifier + "]. Not fetching.");
}
+ numDmeEventsNoData.incrementAndGet();
shuffleManager.addCompletedInputWithNoData(srcAttemptIdentifier);
return;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
index 2cfcc06..99fc18a 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/ShuffleManager.java
@@ -228,11 +228,11 @@ public class ShuffleManager implements FetcherCallback {
ExecutorService fetcherRawExecutor = Executors.newFixedThreadPool(
numFetchers,
new ThreadFactoryBuilder().setDaemon(true)
- .setNameFormat("Fetcher [" + srcNameTrimmed + "] #%d").build());
+ .setNameFormat("Fetcher {" + srcNameTrimmed + "} #%d").build());
this.fetcherExecutor = MoreExecutors.listeningDecorator(fetcherRawExecutor);
ExecutorService schedulerRawExecutor = Executors.newFixedThreadPool(1, new ThreadFactoryBuilder()
- .setDaemon(true).setNameFormat("ShuffleRunner [" + srcNameTrimmed + "]").build());
+ .setDaemon(true).setNameFormat("ShuffleRunner {" + srcNameTrimmed + "}").build());
this.schedulerExecutor = MoreExecutors.listeningDecorator(schedulerRawExecutor);
this.schedulerCallable = new RunShuffleCallable(conf);
@@ -270,7 +270,7 @@ public class ShuffleManager implements FetcherCallback {
shuffleInfoEventsMap = new ConcurrentHashMap<InputIdentifier, ShuffleEventInfo>();
- LOG.info(this.getClass().getSimpleName() + " : numInputs=" + numInputs + ", compressionCodec="
+ LOG.info(srcNameTrimmed + ": numInputs=" + numInputs + ", compressionCodec="
+ (codec == null ? "NoCompressionCodec" : codec.getClass().getName()) + ", numFetchers="
+ numFetchers + ", ifileBufferSize=" + ifileBufferSize + ", ifileReadAheadEnabled="
+ ifileReadAhead + ", ifileReadAheadLength=" + ifileReadAheadLength +", "
@@ -317,7 +317,7 @@ public class ShuffleManager implements FetcherCallback {
}
if (LOG.isDebugEnabled()) {
- LOG.debug("NumCompletedInputs: " + numCompletedInputs);
+ LOG.debug(srcNameTrimmed + ": " + "NumCompletedInputs: " + numCompletedInputs);
}
if (numCompletedInputs.get() < numInputs && !isShutdown.get()) {
lock.lock();
@@ -330,20 +330,20 @@ public class ShuffleManager implements FetcherCallback {
inputHost = pendingHosts.take();
} catch (InterruptedException e) {
if (isShutdown.get()) {
- LOG.info("Interrupted and hasBeenShutdown, Breaking out of ShuffleScheduler Loop");
+ LOG.info(srcNameTrimmed + ": " + "Interrupted and hasBeenShutdown, Breaking out of ShuffleScheduler Loop");
break;
} else {
throw e;
}
}
if (LOG.isDebugEnabled()) {
- LOG.debug("Processing pending host: " + inputHost.toDetailedString());
+ LOG.debug(srcNameTrimmed + ": " + "Processing pending host: " + inputHost.toDetailedString());
}
if (inputHost.getNumPendingInputs() > 0 && !isShutdown.get()) {
Fetcher fetcher = constructFetcherForHost(inputHost, conf);
runningFetchers.add(fetcher);
if (isShutdown.get()) {
- LOG.info("hasBeenShutdown, Breaking out of ShuffleScheduler Loop");
+ LOG.info(srcNameTrimmed + ": " + "hasBeenShutdown, Breaking out of ShuffleScheduler Loop");
}
ListenableFuture<FetchResult> future = fetcherExecutor
.submit(fetcher);
@@ -353,7 +353,7 @@ public class ShuffleManager implements FetcherCallback {
}
} else {
if (LOG.isDebugEnabled()) {
- LOG.debug("Skipping host: " + inputHost.getIdentifier()
+ LOG.debug(srcNameTrimmed + ": " + "Skipping host: " + inputHost.getIdentifier()
+ " since it has no inputs to process");
}
}
@@ -364,8 +364,7 @@ public class ShuffleManager implements FetcherCallback {
}
}
shufflePhaseTime.setValue(System.currentTimeMillis() - startTime);
- LOG.info("Shutting down FetchScheduler, Was Interrupted: " + Thread.currentThread().isInterrupted());
- // TODO NEWTEZ Maybe clean up inputs.
+ LOG.info(srcNameTrimmed + ": " + "Shutting down FetchScheduler, Was Interrupted: " + Thread.currentThread().isInterrupted());
if (!fetcherExecutor.isShutdown()) {
fetcherExecutor.shutdownNow();
}
@@ -450,9 +449,11 @@ public class ShuffleManager implements FetcherCallback {
}
fetcherBuilder.assignWork(inputHost.getHost(), inputHost.getPort(),
inputHost.getSrcPhysicalIndex(), pendingInputsForHost);
- LOG.info("Created Fetcher for host: " + inputHost.getHost()
- + ", info: " + inputHost.getAdditionalInfo()
- + ", with inputs: " + pendingInputsForHost);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Created Fetcher for host: " + inputHost.getHost()
+ + ", info: " + inputHost.getAdditionalInfo()
+ + ", with inputs: " + pendingInputsForHost);
+ }
return fetcherBuilder.build();
}
@@ -471,7 +472,7 @@ public class ShuffleManager implements FetcherCallback {
}
}
if (LOG.isDebugEnabled()) {
- LOG.debug("Adding input: " + srcAttemptIdentifier + ", to host: " + host);
+ LOG.debug(srcNameTrimmed + ": " + "Adding input: " + srcAttemptIdentifier + ", to host: " + host);
}
if (!validateInputAttemptForPipelinedShuffle(srcAttemptIdentifier)) {
@@ -501,7 +502,9 @@ public class ShuffleManager implements FetcherCallback {
public void addCompletedInputWithNoData(
InputAttemptIdentifier srcAttemptIdentifier) {
InputIdentifier inputIdentifier = srcAttemptIdentifier.getInputIdentifier();
- LOG.info("No input data exists for SrcTask: " + inputIdentifier + ". Marking as complete.");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("No input data exists for SrcTask: " + inputIdentifier + ". Marking as complete.");
+ }
if (!completedInputSet.contains(inputIdentifier)) {
synchronized (completedInputSet) {
@@ -574,8 +577,10 @@ public class ShuffleManager implements FetcherCallback {
}
boolean isDone() {
- LOG.info("finalEventId=" + finalEventId + ", eventsProcessed cardinality=" +
- eventsProcessed.cardinality());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("finalEventId=" + finalEventId + ", eventsProcessed cardinality=" +
+ eventsProcessed.cardinality());
+ }
return ((finalEventId != -1) && (finalEventId + 1) == eventsProcessed.cardinality());
}
@@ -631,10 +636,10 @@ public class ShuffleManager implements FetcherCallback {
lock.lock();
try {
totalBytesShuffledTillNow += fetchedBytes;
+ logProgress();
} finally {
lock.unlock();
}
- logProgress();
}
}
}
@@ -751,7 +756,7 @@ public class ShuffleManager implements FetcherCallback {
InputAttemptIdentifier srcAttemptIdentifier, boolean connectFailed) {
// TODO NEWTEZ. Implement logic to report fetch failures after a threshold.
// For now, reporting immediately.
- LOG.info("Fetch failed for src: " + srcAttemptIdentifier
+ LOG.info(srcNameTrimmed + ": " + "Fetch failed for src: " + srcAttemptIdentifier
+ "InputIdentifier: " + srcAttemptIdentifier + ", connectFailed: "
+ connectFailed);
failedShufflesCounter.increment(1);
@@ -885,16 +890,23 @@ public class ShuffleManager implements FetcherCallback {
}
}
+ private final AtomicInteger nextProgressLineEventCount = new AtomicInteger(0);
+
private void logProgress() {
- double mbs = (double) totalBytesShuffledTillNow / (1024 * 1024);
int inputsDone = numCompletedInputs.get();
- long secsSinceStart = (System.currentTimeMillis() - startTime) / 1000 + 1;
- double transferRate = mbs / secsSinceStart;
- LOG.info("copy(" + inputsDone + " (spillsFetched=" + numFetchedSpills.get() + ") of " +
- numInputs +
- ". Transfer rate (CumulativeDataFetched/TimeSinceInputStarted)) "
- + mbpsFormat.format(transferRate) + " MB/s)");
+ if (inputsDone > nextProgressLineEventCount.get() || inputsDone == numInputs) {
+ nextProgressLineEventCount.addAndGet(50);
+ double mbs = (double) totalBytesShuffledTillNow / (1024 * 1024);
+ long secsSinceStart = (System.currentTimeMillis() - startTime) / 1000 + 1;
+
+ double transferRate = mbs / secsSinceStart;
+ LOG.info("copy(" + inputsDone + " (spillsFetched=" + numFetchedSpills.get() + ") of " +
+ numInputs +
+ ". Transfer rate (CumulativeDataFetched/TimeSinceInputStarted)) "
+ + mbpsFormat.format(transferRate) + " MB/s)");
+
+ }
}
@@ -902,15 +914,17 @@ public class ShuffleManager implements FetcherCallback {
@Override
public void onSuccess(Void result) {
- LOG.info("Scheduler thread completed");
+ LOG.info(srcNameTrimmed + ": " + "Scheduler thread completed");
}
@Override
public void onFailure(Throwable t) {
if (isShutdown.get()) {
- LOG.info("Already shutdown. Ignoring error: " + t);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(srcNameTrimmed + ": " + "Already shutdown. Ignoring error: " + t);
+ }
} else {
- LOG.error("Scheduler failed with error: ", t);
+ LOG.error(srcNameTrimmed + ": " + "Scheduler failed with error: ", t);
inputContext.fatalError(t, "Shuffle Scheduler Failed");
}
}
@@ -939,7 +953,9 @@ public class ShuffleManager implements FetcherCallback {
public void onSuccess(FetchResult result) {
fetcher.shutdown();
if (isShutdown.get()) {
- LOG.info("Already shutdown. Ignoring event from fetcher");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(srcNameTrimmed + ": " + "Already shutdown. Ignoring event from fetcher");
+ }
} else {
Iterable<InputAttemptIdentifier> pendingInputs = result.getPendingInputs();
if (pendingInputs != null && pendingInputs.iterator().hasNext()) {
@@ -960,9 +976,11 @@ public class ShuffleManager implements FetcherCallback {
// Unsuccessful - the fetcher may not have shutdown correctly. Try shutting it down.
fetcher.shutdown();
if (isShutdown.get()) {
- LOG.info("Already shutdown. Ignoring error from fetcher: " + t);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(srcNameTrimmed + ": " + "Already shutdown. Ignoring error from fetcher: " + t);
+ }
} else {
- LOG.error("Fetcher failed with error: ", t);
+ LOG.error(srcNameTrimmed + ": " + "Fetcher failed with error: ", t);
shuffleError = t;
inputContext.fatalError(t, "Fetch failed");
doBookKeepingForFetcherComplete();
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java
index 31a8651..604d213 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/SimpleFetchedInputAllocator.java
@@ -60,11 +60,14 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
private final long maxAvailableTaskMemory;
private final long initialMemoryAvailable;
+
+ private final String srcNameTrimmed;
private volatile long usedMemory = 0;
- public SimpleFetchedInputAllocator(String uniqueIdentifier, Configuration conf,
+ public SimpleFetchedInputAllocator(String srcNameTrimmed, String uniqueIdentifier, Configuration conf,
long maxTaskAvailableMemory, long memoryAvailable) {
+ this.srcNameTrimmed = srcNameTrimmed;
this.conf = conf;
this.maxAvailableTaskMemory = maxTaskAvailableMemory;
this.initialMemoryAvailable = memoryAvailable;
@@ -92,8 +95,6 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
this.memoryLimit = initialMemoryAvailable;
}
- LOG.info("RequestedMem=" + memReq + ", Allocated: " + this.memoryLimit);
-
final float singleShuffleMemoryLimitPercent = conf.getFloat(
TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT,
TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT_DEFAULT);
@@ -107,9 +108,13 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
//TODO: cap it to MAX_VALUE until MemoryFetchedInput can handle > 2 GB
this.maxSingleShuffleLimit = (long) Math.min((memoryLimit * singleShuffleMemoryLimitPercent),
Integer.MAX_VALUE);
-
- LOG.info("SimpleInputManager -> " + "MemoryLimit: " +
- this.memoryLimit + ", maxSingleMemLimit: " + this.maxSingleShuffleLimit);
+
+ LOG.info(srcNameTrimmed + ": "
+ + "RequestedMemory=" + memReq
+ + ", AssignedMemorty=" + this.memoryLimit
+ + ", maxSingleShuffleLimit=" + this.maxSingleShuffleLimit
+ );
+
}
@Private
@@ -137,7 +142,10 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
fileNameAllocator);
} else {
this.usedMemory += actualSize;
- LOG.info("Used memory after allocating " + actualSize + " : " + usedMemory);
+ if (LOG.isDebugEnabled()) {
+ LOG.info(srcNameTrimmed + ": " + "Used memory after allocating " + actualSize + " : " +
+ usedMemory);
+ }
return new MemoryFetchedInput(actualSize, compressedSize, inputAttemptIdentifier, this);
}
}
@@ -196,7 +204,9 @@ public class SimpleFetchedInputAllocator implements FetchedInputAllocator,
private synchronized void unreserve(long size) {
this.usedMemory -= size;
- LOG.info("Used memory after freeing " + size + " : " + usedMemory);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(srcNameTrimmed + ": " + "Used memory after freeing " + size + " : " + usedMemory);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java
index 67b5aa0..0ba37dd 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/FetcherOrderedGrouped.java
@@ -90,6 +90,8 @@ class FetcherOrderedGrouped extends Thread {
volatile HttpURLConnection connection;
volatile DataInputStream input;
+ volatile MapHost assignedHost = null;
+
HttpConnection httpConnection;
HttpConnectionParams httpConnectionParams;
@@ -139,15 +141,15 @@ class FetcherOrderedGrouped extends Thread {
this.localDiskFetchEnabled = localDiskFetchEnabled;
- this.logIdentifier = "fetcher [" + TezUtilsInternal
- .cleanVertexName(inputContext.getSourceVertexName()) + "] #" + id;
+ this.logIdentifier = "fetcher {" + TezUtilsInternal
+ .cleanVertexName(inputContext.getSourceVertexName()) + "} #" + id;
setName(logIdentifier);
setDaemon(true);
}
@VisibleForTesting
protected void fetchNext() throws InterruptedException, IOException {
- MapHost host = null;
+ assignedHost = null;
try {
// If merge is on, block
merger.waitForInMemoryMerge();
@@ -156,20 +158,20 @@ class FetcherOrderedGrouped extends Thread {
merger.waitForShuffleToMergeMemory();
// Get a host to shuffle from
- host = scheduler.getHost();
+ assignedHost = scheduler.getHost();
metrics.threadBusy();
- String hostPort = host.getHostIdentifier();
+ String hostPort = assignedHost.getHostIdentifier();
if (localDiskFetchEnabled && hostPort.equals(localShuffleHostPort)) {
- setupLocalDiskFetch(host);
+ setupLocalDiskFetch(assignedHost);
} else {
// Shuffle
- copyFromHost(host);
+ copyFromHost(assignedHost);
}
} finally {
cleanupCurrentConnection(false);
- if (host != null) {
- scheduler.freeHost(host);
+ if (assignedHost != null) {
+ scheduler.freeHost(assignedHost);
metrics.threadFree();
}
}
@@ -191,6 +193,9 @@ class FetcherOrderedGrouped extends Thread {
public void shutDown() throws InterruptedException {
this.stopped = true;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Fetcher stopped for host " + assignedHost);
+ }
interrupt();
cleanupCurrentConnection(true);
try {
@@ -276,14 +281,19 @@ class FetcherOrderedGrouped extends Thread {
// Setup connection again if disconnected
cleanupCurrentConnection(true);
if (stopped) {
- LOG.info("Not re-establishing connection since Fetcher has been stopped");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Not re-establishing connection since Fetcher has been stopped");
+ }
return;
}
// Connect with retry
if (!setupConnection(host, new LinkedList<InputAttemptIdentifier>(remaining))) {
if (stopped) {
cleanupCurrentConnection(true);
- LOG.info("Not reporting connection re-establishment failure since fetcher is stopped");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Not reporting connection re-establishment failure since fetcher is stopped");
+ }
return;
}
failedTasks = new InputAttemptIdentifier[] {getNextRemainingAttempt()};
@@ -294,8 +304,10 @@ class FetcherOrderedGrouped extends Thread {
if (failedTasks != null && failedTasks.length > 0) {
if (stopped) {
- LOG.info("Ignoring copyMapOutput failures for tasks: " + Arrays.toString(failedTasks) +
- " since Fetcher has been stopped");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ignoring copyMapOutput failures for tasks: " + Arrays.toString(failedTasks) +
+ " since Fetcher has been stopped");
+ }
} else {
LOG.warn("copyMapOutput failed for tasks " + Arrays.toString(failedTasks));
for (InputAttemptIdentifier left : failedTasks) {
@@ -328,7 +340,9 @@ class FetcherOrderedGrouped extends Thread {
connectSucceeded = httpConnection.connect();
if (stopped) {
- LOG.info("Detected fetcher has been shutdown after connection establishment. Returning");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Detected fetcher has been shutdown after connection establishment. Returning");
+ }
return false;
}
input = httpConnection.getInputStream();
@@ -336,7 +350,9 @@ class FetcherOrderedGrouped extends Thread {
return true;
} catch (IOException ie) {
if (stopped) {
- LOG.info("Not reporting fetch failure, since an Exception was caught after shutdown");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Not reporting fetch failure, since an Exception was caught after shutdown");
+ }
return false;
}
ioErrs.increment(1);
@@ -402,7 +418,9 @@ class FetcherOrderedGrouped extends Thread {
InputAttemptIdentifier.PATH_PREFIX + ", partition: " + header.forReduce);
return new InputAttemptIdentifier[] {getNextRemainingAttempt()};
} else {
- LOG.info("Already shutdown. Ignoring invalid map id error");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Already shutdown. Ignoring invalid map id error");
+ }
return EMPTY_ATTEMPT_ID_ARRAY;
}
}
@@ -419,8 +437,10 @@ class FetcherOrderedGrouped extends Thread {
// the remaining because we dont know where to start reading from. YARN-1773
return new InputAttemptIdentifier[] {getNextRemainingAttempt()};
} else {
- LOG.info("Already shutdown. Ignoring invalid map id error. Exception: " +
- e.getClass().getName() + ", Message: " + e.getMessage());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Already shutdown. Ignoring invalid map id error. Exception: " +
+ e.getClass().getName() + ", Message: " + e.getMessage());
+ }
return EMPTY_ATTEMPT_ID_ARRAY;
}
}
@@ -436,7 +456,9 @@ class FetcherOrderedGrouped extends Thread {
assert (srcAttemptId != null);
return new InputAttemptIdentifier[]{srcAttemptId};
} else {
- LOG.info("Already stopped. Ignoring verification failure.");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Already stopped. Ignoring verification failure.");
+ }
return EMPTY_ATTEMPT_ID_ARRAY;
}
}
@@ -455,7 +477,9 @@ class FetcherOrderedGrouped extends Thread {
ioErrs.increment(1);
scheduler.reportLocalError(e);
} else {
- LOG.info("Already stopped. Ignoring error from merger.reserve");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Already stopped. Ignoring error from merger.reserve");
+ }
}
return EMPTY_ATTEMPT_ID_ARRAY;
}
@@ -468,16 +492,19 @@ class FetcherOrderedGrouped extends Thread {
}
// Go!
- LOG.info("fetcher#" + id + " about to shuffle output of map " +
- mapOutput.getAttemptIdentifier() + " decomp: " +
- decompressedLength + " len: " + compressedLength + " to " + mapOutput.getType());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("fetcher#" + id + " about to shuffle output of map " +
+ mapOutput.getAttemptIdentifier() + " decomp: " +
+ decompressedLength + " len: " + compressedLength + " to " + mapOutput.getType());
+ }
+
if (mapOutput.getType() == Type.MEMORY) {
ShuffleUtils.shuffleToMemory(mapOutput.getMemory(), input,
(int) decompressedLength, (int) compressedLength, codec, ifileReadAhead,
ifileReadAheadLength, LOG, mapOutput.getAttemptIdentifier().toString());
} else if (mapOutput.getType() == Type.DISK) {
ShuffleUtils.shuffleToDisk(mapOutput.getDisk(), host.getHostIdentifier(),
- input, compressedLength, LOG, mapOutput.getAttemptIdentifier().toString());
+ input, compressedLength, decompressedLength, LOG, mapOutput.getAttemptIdentifier().toString());
} else {
throw new IOException("Unknown mapOutput type while fetching shuffle data:" +
mapOutput.getType());
@@ -496,8 +523,10 @@ class FetcherOrderedGrouped extends Thread {
return null;
} catch (IOException ioe) {
if (stopped) {
- LOG.info("Not reporting fetch failure for exception during data copy: ["
- + ioe.getClass().getName() + ", " + ioe.getMessage() + "]");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Not reporting fetch failure for exception during data copy: ["
+ + ioe.getClass().getName() + ", " + ioe.getMessage() + "]");
+ }
cleanupCurrentConnection(true);
if (mapOutput != null) {
mapOutput.abort(); // Release resources
@@ -666,7 +695,10 @@ class FetcherOrderedGrouped extends Thread {
LOG.warn("Failed to read local disk output of " + srcAttemptId + " from " +
host.getHostIdentifier(), e);
} else {
- LOG.info("Ignoring fetch error during local disk copy since fetcher has already been stopped");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Ignoring fetch error during local disk copy since fetcher has already been stopped");
+ }
return;
}
}
http://git-wip-us.apache.org/repos/asf/tez/blob/cd80d9ab/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
index 2e6ebd9..e25c064 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/MergeManager.java
@@ -58,14 +58,13 @@ import org.apache.tez.runtime.library.common.Constants;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
import org.apache.tez.runtime.library.common.combine.Combiner;
import org.apache.tez.runtime.library.common.sort.impl.IFile;
-import org.apache.tez.runtime.library.common.sort.impl.TezMerger;
-import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
import org.apache.tez.runtime.library.common.sort.impl.IFile.Writer;
+import org.apache.tez.runtime.library.common.sort.impl.TezMerger;
import org.apache.tez.runtime.library.common.sort.impl.TezMerger.Segment;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
import org.apache.tez.runtime.library.common.task.local.output.TezTaskOutputFiles;
import org.apache.tez.runtime.library.hadoop.compat.NullProgressable;
-
/**
* Usage. Create instance. setInitialMemoryAvailable(long), configureAndStart()
*
@@ -219,10 +218,15 @@ public class MergeManager {
} else {
this.postMergeMemLimit = maxRedBuffer;
}
-
- LOG.info("InitialRequest: ShuffleMem=" + memLimit + ", postMergeMem=" + maxRedBuffer
- + ", RuntimeTotalAvailable=" + this.initialMemoryAvailable + ". Updated to: ShuffleMem="
- + this.memoryLimit + ", postMergeMem: " + this.postMergeMemLimit);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ inputContext.getSourceVertexName() + ": " + "InitialRequest: ShuffleMem=" + memLimit +
+ ", postMergeMem=" + maxRedBuffer
+ + ", RuntimeTotalAvailable=" + this.initialMemoryAvailable +
+ ". Updated to: ShuffleMem="
+ + this.memoryLimit + ", postMergeMem: " + this.postMergeMemLimit);
+ }
this.ioSortFactor =
conf.getInt(
@@ -252,10 +256,11 @@ public class MergeManager {
conf.getFloat(
TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT,
TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT_DEFAULT));
- LOG.info("MergerManager: memoryLimit=" + memoryLimit + ", " +
+ LOG.info(inputContext.getSourceVertexName() + ": MergerManager: memoryLimit=" + memoryLimit + ", " +
"maxSingleShuffleLimit=" + maxSingleShuffleLimit + ", " +
"mergeThreshold=" + mergeThreshold + ", " +
"ioSortFactor=" + ioSortFactor + ", " +
+ "postMergeMem=" + postMergeMemLimit + ", " +
"memToMemMergeOutputsThreshold=" + memToMemMergeOutputsThreshold);
if (this.maxSingleShuffleLimit >= this.mergeThreshold) {
@@ -310,8 +315,6 @@ public class MergeManager {
long memLimit = conf.getLong(Constants.TEZ_RUNTIME_TASK_MEMORY,
(long)(maxAvailableTaskMemory * maxInMemCopyUse));
- LOG.info("Initial Shuffle Memory Required: " + memLimit + ", based on INPUT_BUFFER_factor: " + maxInMemCopyUse);
-
float maxRedPer = conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_POST_MERGE_BUFFER_PERCENT,
TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT_DEFAULT);
if (maxRedPer > 1.0 || maxRedPer < 0.0) {
@@ -319,7 +322,9 @@ public class MergeManager {
}
long maxRedBuffer = (long) (maxAvailableTaskMemory * maxRedPer);
- LOG.info("Initial Memory required for final merged output: " + maxRedBuffer + ", using factor: " + maxRedPer);
+ LOG.info("Initial Memory required for SHUFFLE_BUFFER=" + memLimit +
+ " based on INPUT_BUFFER_FACTOR=" + maxInMemCopyUse + ", for final merged output=" +
+ maxRedBuffer + ", using factor: " + maxRedPer);
long reqMem = Math.max(maxRedBuffer, memLimit);
return reqMem;
@@ -373,9 +378,11 @@ public class MergeManager {
int fetcher
) throws IOException {
if (!canShuffleToMemory(requestedSize)) {
- LOG.info(srcAttemptIdentifier + ": Shuffling to disk since " + requestedSize +
- " is greater than maxSingleShuffleLimit (" +
- maxSingleShuffleLimit + ")");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(srcAttemptIdentifier + ": Shuffling to disk since " + requestedSize +
+ " is greater than maxSingleShuffleLimit (" +
+ maxSingleShuffleLimit + ")");
+ }
return MapOutput.createDiskMapOutput(srcAttemptIdentifier, this, compressedLength, conf,
fetcher, true, mapOutputFile);
}
@@ -437,9 +444,9 @@ public class MergeManager {
public synchronized void closeInMemoryFile(MapOutput mapOutput) {
inMemoryMapOutputs.add(mapOutput);
LOG.info("closeInMemoryFile -> map-output of size: " + mapOutput.getSize()
- + ", inMemoryMapOutputs.size() -> " + inMemoryMapOutputs.size()
- + ", commitMemory -> " + commitMemory + ", usedMemory ->" + usedMemory + ", mapOutput=" +
- mapOutput);
+ + ", inMemoryMapOutputs.size() -> " + inMemoryMapOutputs.size()
+ + ", commitMemory -> " + commitMemory + ", usedMemory ->" + usedMemory + ", mapOutput=" +
+ mapOutput);
commitMemory+= mapOutput.getSize();
@@ -461,7 +468,7 @@ public class MergeManager {
private void startMemToDiskMerge() {
synchronized (inMemoryMerger) {
if (!inMemoryMerger.isInProgress()) {
- LOG.info("Starting inMemoryMerger's merge since commitMemory=" +
+ LOG.info(inputContext.getSourceVertexName() + ": " + "Starting inMemoryMerger's merge since commitMemory=" +
commitMemory + " > mergeThreshold=" + mergeThreshold +
". Current usedMemory=" + usedMemory);
inMemoryMapOutputs.addAll(inMemoryMergedMapOutputs);
@@ -473,7 +480,7 @@ public class MergeManager {
public synchronized void closeInMemoryMergedFile(MapOutput mapOutput) {
inMemoryMergedMapOutputs.add(mapOutput);
- LOG.info("closeInMemoryMergedFile -> size: " + mapOutput.getSize() +
+ LOG.info("closeInMemoryMergedFile -> size: " + mapOutput.getSize() +
", inMemoryMergedMapOutputs.size() -> " +
inMemoryMergedMapOutputs.size());
}
@@ -573,7 +580,7 @@ public class MergeManager {
Writer writer =
new InMemoryWriter(mergedMapOutputs.getArrayStream());
- LOG.info("Initiating Memory-to-Memory merge with " + noInMemorySegments +
+ LOG.info(inputContext.getSourceVertexName() + ": " + "Initiating Memory-to-Memory merge with " + noInMemorySegments +
" segments of total-size: " + mergeOutputSize);
// Nothing will be materialized to disk because the sort factor is being
@@ -590,7 +597,7 @@ public class MergeManager {
TezMerger.writeFile(rIter, writer, nullProgressable, TezRuntimeConfiguration.TEZ_RUNTIME_RECORDS_BEFORE_PROGRESS_DEFAULT);
writer.close();
- LOG.info(inputContext.getUniqueIdentifier() +
+ LOG.info(inputContext.getSourceVertexName() +
" Memory-to-Memory merge of the " + noInMemorySegments +
" files in-memory complete.");
@@ -642,7 +649,6 @@ public class MergeManager {
Path outputPath = mapOutputFile.getInputFileForWrite(
srcTaskIdentifier.getInputIdentifier().getInputIndex(), srcTaskIdentifier.getSpillEventId(),
mergeOutputSize).suffix(Constants.MERGED_OUTPUT_PREFIX);
- LOG.info("Patch..InMemoryMerger outputPath: " + outputPath);
Writer writer = null;
long outFileLen = 0;
@@ -801,7 +807,7 @@ public class MergeManager {
final long outputLen = localFS.getFileStatus(outputPath).getLen();
closeOnDiskFile(new FileChunk(outputPath, 0, outputLen));
- LOG.info(inputContext.getUniqueIdentifier() +
+ LOG.info(inputContext.getSourceVertexName() +
" Finished merging " + inputs.size() +
" map output files on disk of total-size " +
approxOutputSize + "." +