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/21 19:36:45 UTC
[1/2] tez git commit: TEZ-2775. Improve and consolidate logging in
Runtime components. Contributed by Siddharth Seth and Rajesh Balamohan.
Repository: tez
Updated Branches:
refs/heads/master 7ed7025ad -> 983ceeee1
http://git-wip-us.apache.org/repos/asf/tez/blob/983ceeee/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 18c8302..0e1fe9f 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
@@ -144,7 +144,7 @@ public class Shuffle implements ExceptionReporter {
TezCounter mergedMapOutputsCounter =
inputContext.getCounters().findCounter(TaskCounter.MERGED_MAP_OUTPUTS);
- 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);
@@ -190,7 +190,7 @@ 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());
executor = MoreExecutors.listeningDecorator(rawExecutor);
@@ -201,7 +201,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());
}
}
@@ -327,7 +327,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;
}
}
@@ -337,7 +337,7 @@ public class Shuffle implements ExceptionReporter {
cleanupShuffleScheduler();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- 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");
}
}
@@ -355,13 +355,13 @@ public class Shuffle implements ExceptionReporter {
if (ignoreErrors) {
//Reset the status
Thread.currentThread().interrupt();
- LOG.info("Interrupted while attempting to close the merger during cleanup. Ignoring");
+ LOG.info(srcNameTrimmed + ": " + "Interrupted while attempting to close the merger during cleanup. Ignoring");
} else {
throw e;
}
} 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;
}
@@ -371,10 +371,13 @@ public class Shuffle implements ExceptionReporter {
private void cleanupIgnoreErrors() {
try {
+ if (eventHandler != null) {
+ eventHandler.logProgress(true);
+ }
cleanupShuffleSchedulerIgnoreErrors();
cleanupMerger(true);
} catch (Throwable t) {
- LOG.info("Error in cleaning up.., ", t);
+ LOG.info(srcNameTrimmed + ": " + "Error in cleaning up.., ", t);
}
}
@@ -383,7 +386,7 @@ 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("Setting throwable in reportException with message [" + t.getMessage() +
+ LOG.info(srcNameTrimmed + ": " + "Setting throwable in reportException with message [" + t.getMessage() +
"] from thread [" + Thread.currentThread().getName());
throwable.set(t);
throwingThreadName = Thread.currentThread().getName();
@@ -409,15 +412,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");
+ LOG.info(srcNameTrimmed + ": " + "Already shutdown. Ignoring error");
} else {
- LOG.error("ShuffleRunner failed with error", t);
+ LOG.error(srcNameTrimmed + ": " + "ShuffleRunner failed with error", t);
// In case of an abort / Interrupt - the runtime makes sure that this is ignored.
inputContext.fatalError(t, "Shuffle Runner Failed");
cleanupIgnoreErrors();
http://git-wip-us.apache.org/repos/asf/tez/blob/983ceeee/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 9481e65..e0473b3 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/983ceeee/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 26464bb..f45ca35 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
@@ -207,7 +207,6 @@ class ShuffleScheduler {
TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES,
TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES_DEFAULT);
numFetchers = Math.min(configuredNumFetchers, numInputs);
- LOG.info("Num fetchers determined to be: " + numFetchers);
localDiskFetchEnabled = conf.getBoolean(
TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH,
@@ -260,7 +259,7 @@ class ShuffleScheduler {
ExecutorService fetcherRawExecutor = Executors.newFixedThreadPool(numFetchers,
new ThreadFactoryBuilder().setDaemon(true)
- .setNameFormat("Fetcher [" + srcNameTrimmed + "] #%d").build());
+ .setNameFormat("Fetcher {" + srcNameTrimmed + "} #%d").build());
this.fetcherExecutor = MoreExecutors.listeningDecorator(fetcherRawExecutor);
this.maxFailedUniqueFetches = Math.min(numberOfInputs, 5);
@@ -293,7 +292,8 @@ class ShuffleScheduler {
+ ", maxFailedUniqueFetches=" + maxFailedUniqueFetches
+ ", abortFailureLimit=" + abortFailureLimit
+ ", maxMapRuntime=" + maxMapRuntime
- + ", maxTaskOutputAtOnce=" + maxTaskOutputAtOnce);
+ + ", maxTaskOutputAtOnce=" + maxTaskOutputAtOnce
+ + ", numFetchers=" + numFetchers);
}
public void start() throws Exception {
@@ -305,6 +305,8 @@ class ShuffleScheduler {
public void close() throws InterruptedException {
if (!isShutdown.getAndSet(true)) {
+ logProgress();
+
// Notify and interrupt the waiting scheduler thread
synchronized (this) {
notifyAll();
@@ -518,19 +520,24 @@ class ShuffleScheduler {
@VisibleForTesting
void reportExceptionForInput(Exception exception) {
- LOG.error("Reporting exception for input", exception);
+ LOG.error(srcNameTrimmed + ": " + "Reporting exception for input", exception);
exceptionReporter.reportException(exception);
}
+ private final AtomicInteger nextProgressLineEventCount = new AtomicInteger(0);
+
private void logProgress() {
- double mbs = (double) totalBytesShuffledTillNow / (1024 * 1024);
int inputsDone = numInputs - remainingMaps.get();
- 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 || isShutdown.get()) {
+ 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,
@@ -585,7 +592,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
exceptionReporter.reportException(ioe);
}
@@ -598,7 +605,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(),
@@ -655,7 +662,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);
@@ -700,7 +708,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.
@@ -719,7 +727,9 @@ class ShuffleScheduler {
public synchronized MapHost getHost() throws InterruptedException {
while (pendingHosts.isEmpty() && remainingMaps.get() > 0) {
- LOG.info("PendingHosts=" + pendingHosts);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("PendingHosts=" + pendingHosts);
+ }
wait();
}
@@ -735,7 +745,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());
@@ -844,8 +854,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() {
@@ -896,8 +908,8 @@ class ShuffleScheduler {
*/
private class Referee extends Thread {
public Referee() {
- setName("ShufflePenaltyReferee ["
- + TezUtilsInternal.cleanVertexName(inputContext.getSourceVertexName()) + "]");
+ setName("ShufflePenaltyReferee {"
+ + TezUtilsInternal.cleanVertexName(inputContext.getSourceVertexName()) + "}");
setDaemon(true);
}
@@ -955,7 +967,7 @@ class ShuffleScheduler {
ShuffleScheduler.this.wait();
} catch (InterruptedException e) {
if (isShutdown.get()) {
- LOG.info(
+ LOG.info(srcNameTrimmed + ": " +
"Interrupted while waiting for fetchers to complete and hasBeenShutdown. Breaking out of ShuffleSchedulerCallable loop");
Thread.currentThread().interrupt();
break;
@@ -968,7 +980,7 @@ class ShuffleScheduler {
}
if (LOG.isDebugEnabled()) {
- LOG.debug("NumCompletedInputs: {}" + (numInputs - remainingMaps.get()));
+ LOG.debug(srcNameTrimmed + ": " + "NumCompletedInputs: {}" + (numInputs - remainingMaps.get()));
}
// Ensure there's memory available before scheduling the next Fetcher.
@@ -979,7 +991,7 @@ class ShuffleScheduler {
mergeManager.waitForShuffleToMergeMemory();
} catch (InterruptedException e) {
if (isShutdown.get()) {
- LOG.info(
+ LOG.info(srcNameTrimmed + ": " +
"Interrupted while waiting for merge to complete and hasBeenShutdown. Breaking out of ShuffleSchedulerCallable loop");
Thread.currentThread().interrupt();
break;
@@ -998,7 +1010,7 @@ class ShuffleScheduler {
mapHost = getHost(); // Leads to a wait.
} catch (InterruptedException e) {
if (isShutdown.get()) {
- LOG.info(
+ LOG.info(srcNameTrimmed + ": " +
"Interrupted while waiting for host and hasBeenShutdown. Breaking out of ShuffleSchedulerCallable loop");
Thread.currentThread().interrupt();
break;
@@ -1010,11 +1022,14 @@ class ShuffleScheduler {
break; // Check for the exit condition.
}
if (LOG.isDebugEnabled()) {
- LOG.debug("Processing pending host: " + mapHost.toString());
+ LOG.debug(srcNameTrimmed + ": " + "Processing pending host: " + mapHost.toString());
}
if (!isShutdown.get()) {
count++;
- LOG.info("Scheduling fetch for inputHost: {}", mapHost.getIdentifier());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(srcNameTrimmed + ": " + "Scheduling fetch for inputHost: {}",
+ mapHost.getIdentifier());
+ }
FetcherOrderedGrouped fetcherOrderedGrouped = constructFetcherForHost(mapHost);
runningFetchers.add(fetcherOrderedGrouped);
ListenableFuture<Void> future = fetcherExecutor.submit(fetcherOrderedGrouped);
@@ -1063,7 +1078,7 @@ class ShuffleScheduler {
public void onSuccess(Void result) {
fetcherOrderedGrouped.shutDown();
if (isShutdown.get()) {
- LOG.info("Already shutdown. Ignoring fetch complete");
+ LOG.info(srcNameTrimmed + ": " + "Already shutdown. Ignoring fetch complete");
} else {
doBookKeepingForFetcherComplete();
}
@@ -1073,9 +1088,9 @@ class ShuffleScheduler {
public void onFailure(Throwable t) {
fetcherOrderedGrouped.shutDown();
if (isShutdown.get()) {
- LOG.info("Already shutdown. Ignoring fetch complete");
+ LOG.info(srcNameTrimmed + ": " + "Already shutdown. Ignoring fetch complete");
} else {
- LOG.error("Fetcher failed with error", t);
+ LOG.error(srcNameTrimmed + ": " + "Fetcher failed with error", t);
exceptionReporter.reportException(t);
doBookKeepingForFetcherComplete();
}
http://git-wip-us.apache.org/repos/asf/tez/blob/983ceeee/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 d27d08d..aa521ea 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
@@ -169,7 +169,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;
@@ -187,9 +190,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);
@@ -244,8 +251,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);
@@ -339,9 +345,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/983ceeee/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 c4b2b3d..81f5211 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
@@ -120,7 +120,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
@@ -140,10 +142,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);
@@ -157,7 +178,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
@@ -167,17 +187,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,7 +231,9 @@ public class PipelinedSorter extends ExternalSorter {
merger.add(span.sort(sorter));
boolean ret = spill(true);
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 && ret) {
sendPipelinedShuffleEvents();
}
@@ -257,7 +273,8 @@ public class PipelinedSorter extends ExternalSorter {
(numSpills - 1), indexCacheList.get(numSpills - 1), partitions, sendEmptyPartitionDetails,
pathComponent, partitionStats);
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
@@ -363,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) {
if (isThreadInterrupted()) {
return;
@@ -403,10 +424,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
@@ -437,8 +454,8 @@ public class PipelinedSorter extends ExternalSorter {
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- LOG.info("Interrupted while waiting for mergers to complete");
- throw new IOInterruptedException("Interrupted while waiting for mergers to complete", e);
+ LOG.info(outputContext.getDestinationVertexName() + ": Interrupted while waiting for mergers to complete");
+ throw new IOInterruptedException(outputContext.getDestinationVertexName() + ": Interrupted while waiting for mergers to complete", e);
}
// create spill file
@@ -449,7 +466,7 @@ public class PipelinedSorter extends ExternalSorter {
mapOutputFile.getSpillFileForWrite(numSpills, size);
spillFilePaths.put(numSpills, filename);
out = rfs.create(filename, true, 4096);
- LOG.info("Spilling to " + filename.toString());
+ LOG.info(outputContext.getDestinationVertexName() + ": Spilling to " + filename.toString());
for (int i = 0; i < partitions; ++i) {
if (isThreadInterrupted()) {
return false;
@@ -509,7 +526,7 @@ public class PipelinedSorter extends ExternalSorter {
cleanup();
}
sortmaster.shutdownNow();
- LOG.info("Thread interrupted, cleaned up stale data, sorter threads shutdown=" + sortmaster
+ LOG.info(outputContext.getDestinationVertexName() + ": Thread interrupted, cleaned up stale data, sorter threads shutdown=" + sortmaster
.isShutdown() + ", terminated=" + sortmaster.isTerminated());
return true;
}
@@ -530,7 +547,7 @@ public class PipelinedSorter extends ExternalSorter {
}
try {
- LOG.info("Starting flush of map output");
+ LOG.info(outputContext.getDestinationVertexName() + ": Starting flush of map output");
span.end();
merger.add(span.sort(sorter));
// force a spill in flush()
@@ -549,7 +566,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;
}
@@ -567,7 +591,7 @@ public class PipelinedSorter extends ExternalSorter {
ShuffleUtils.generateEventOnSpill(events, isFinalMergeEnabled(), isLastEvent,
outputContext, i, indexCacheList.get(i), partitions,
sendEmptyPartitionDetails, pathComponent, partitionStats);
- 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;
@@ -586,8 +610,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);
}
@@ -608,7 +633,7 @@ public class PipelinedSorter extends ExternalSorter {
mapOutputFile.getOutputIndexFileForWrite(0); //TODO
if (LOG.isDebugEnabled()) {
- LOG.debug(
+ LOG.debug(outputContext.getDestinationVertexName() + ": " +
"numSpills: " + numSpills + ", finalOutputFile:" + finalOutputFile + ", finalIndexFile:"
+ finalIndexFile);
}
@@ -773,8 +798,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();
@@ -793,7 +818,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);
}
@@ -861,7 +886,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;
}
@@ -883,11 +908,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();
@@ -1184,10 +1209,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(ExecutionException e) {
- LOG.info(e.toString());
+ LOG.info(outputContext.getDestinationVertexName() + ": " + e.toString());
return false;
}
}
http://git-wip-us.apache.org/repos/asf/tez/blob/983ceeee/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 edc02f3..727f9a2 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
@@ -130,7 +130,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
@@ -144,7 +144,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.");
}
@@ -164,10 +165,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
@@ -177,8 +182,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();
@@ -200,7 +205,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 +
@@ -208,7 +213,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)");
}
@@ -354,7 +359,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;
@@ -373,7 +378,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) + ")");
}
}
@@ -391,7 +396,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) + ")");
}
}
@@ -647,7 +652,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
spillThread.interrupt();
spillThread.join();
} catch (InterruptedException e) {
- LOG.info("Spill thread interrupted");
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "Spill thread interrupted");
//Reset status
Thread.currentThread().interrupt();
throw new IOInterruptedException("Spill failed", e);
@@ -656,7 +661,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
@Override
public void flush() throws IOException {
- LOG.info("Starting flush of map output");
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "Starting flush of map output");
if (Thread.currentThread().isInterrupted()) {
/**
* Possible that the thread got interrupted when flush was happening or when the flush was
@@ -691,13 +696,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();
}
@@ -743,7 +748,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();
@@ -756,7 +761,7 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
}
}
} catch (InterruptedException e) {
- LOG.info("Spill thread interrupted");
+ LOG.info(outputContext.getDestinationVertexName() + ": " + "Spill thread interrupted");
Thread.currentThread().interrupt();
} finally {
spillLock.unlock();
@@ -787,13 +792,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();
}
@@ -889,7 +892,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);
}
@@ -927,7 +930,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);
@@ -1113,7 +1116,8 @@ public class DefaultSorter extends ExternalSorter implements IndexedSortable {
outputContext, index, spillRecord, partitions, sendEmptyPartitionDetails, pathComponent,
partitionStats);
- 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);
@@ -1271,7 +1275,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/983ceeee/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 9a98cd1..70b345f 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
@@ -92,6 +92,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;
@@ -150,6 +151,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
@@ -171,13 +173,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);
@@ -190,8 +195,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];
@@ -214,7 +219,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) {
@@ -321,7 +331,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();
@@ -420,10 +430,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;
@@ -461,7 +471,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();
}
@@ -469,7 +479,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();
@@ -480,7 +490,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();
@@ -687,7 +697,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);
@@ -739,7 +749,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)
@@ -791,9 +801,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);
}
@@ -902,10 +912,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");
}
}
@@ -925,7 +935,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);
@@ -935,7 +944,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");
}
@@ -960,7 +969,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/983ceeee/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 7d887de..880dc2f 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
@@ -132,8 +132,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);
@@ -283,10 +285,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/983ceeee/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 271eed3..fad164f 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.apache.tez.runtime.library.common.Constants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -132,7 +133,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());
@@ -150,8 +152,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);
@@ -208,6 +212,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();
}
@@ -218,7 +226,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/983ceeee/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 98d75a9..45b6713 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
@@ -126,7 +126,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);
@@ -184,8 +184,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/983ceeee/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 b50f17d..879c2e0 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/983ceeee/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 7498627..90c0ed4 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/983ceeee/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/983ceeee/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 c22e605..b531464 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
@@ -214,19 +214,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-2775. Improve and consolidate logging in
Runtime components. Contributed by Siddharth Seth and Rajesh Balamohan.
Posted by ss...@apache.org.
TEZ-2775. Improve and consolidate logging in Runtime components.
Contributed by Siddharth Seth and Rajesh Balamohan.
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/983ceeee
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/983ceeee
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/983ceeee
Branch: refs/heads/master
Commit: 983ceeee19535b92e60b1f8c61bb1e8786b8ffa0
Parents: 7ed7025
Author: Siddharth Seth <ss...@apache.org>
Authored: Mon Sep 21 10:36:32 2015 -0700
Committer: Siddharth Seth <ss...@apache.org>
Committed: Mon Sep 21 10:36:32 2015 -0700
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../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 +-
.../org/apache/tez/http/HttpConnection.java | 4 +-
.../library/common/InputAttemptIdentifier.java | 2 +-
.../runtime/library/common/TezRuntimeUtils.java | 8 +-
.../runtime/library/common/ValuesIterator.java | 2 -
.../runtime/library/common/shuffle/Fetcher.java | 109 +++++++++++++------
.../common/shuffle/ShuffleEventHandler.java | 1 +
.../library/common/shuffle/ShuffleUtils.java | 40 +++++--
.../impl/ShuffleInputEventHandlerImpl.java | 41 +++++--
.../common/shuffle/impl/ShuffleManager.java | 84 ++++++++------
.../impl/SimpleFetchedInputAllocator.java | 26 +++--
.../orderedgrouped/FetcherOrderedGrouped.java | 66 ++++++++---
.../shuffle/orderedgrouped/MergeManager.java | 81 +++++++-------
.../common/shuffle/orderedgrouped/Shuffle.java | 27 +++--
.../ShuffleInputEventHandlerOrderedGrouped.java | 42 +++++--
.../orderedgrouped/ShuffleScheduler.java | 75 ++++++++-----
.../common/sort/impl/ExternalSorter.java | 26 +++--
.../common/sort/impl/PipelinedSorter.java | 107 +++++++++++-------
.../common/sort/impl/dflt/DefaultSorter.java | 75 +++++++------
.../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 +-
.../sort/impl/dflt/TestDefaultSorter.java | 10 +-
36 files changed, 713 insertions(+), 364 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/983ceeee/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 0e93702..ee1e255 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -7,6 +7,7 @@ Release 0.8.1: Unreleased
INCOMPATIBLE CHANGES
ALL CHANGES:
+ TEZ-2775. Improve and consolidate logging in Runtime components.
TEZ-2097. TEZ-UI Add dag logs backend support
TEZ-2812. Preemption sometimes does not respect heartbeats between preemptions
TEZ-814. Improve heuristic for determining a task has failed outputs
@@ -185,6 +186,7 @@ Release 0.7.1: Unreleased
INCOMPATIBLE CHANGES
ALL CHANGES:
+ TEZ-2775. Improve and consolidate logging in Runtime components.
TEZ-2097. TEZ-UI Add dag logs backend support
TEZ-2812. Preemption sometimes does not respect heartbeats between preemptions
TEZ-814. Improve heuristic for determining a task has failed outputs
http://git-wip-us.apache.org/repos/asf/tez/blob/983ceeee/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/983ceeee/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/983ceeee/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/983ceeee/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/983ceeee/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/983ceeee/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/983ceeee/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 a3b19ed..12e5092 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
@@ -344,7 +344,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);
@@ -381,6 +380,8 @@ public class MROutput extends AbstractLogicalOutput {
}
}
+ String outputFormatClassName;
+
outputRecordCounter = getContext().getCounters().findCounter(TaskCounter.OUTPUT_RECORDS);
if (useNewApi) {
@@ -389,6 +390,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);
}
@@ -405,6 +407,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();
@@ -415,8 +418,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;
}
@@ -518,6 +522,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);
@@ -535,7 +540,6 @@ public class MROutput extends AbstractLogicalOutput {
return;
}
- LOG.info("Flushing Simple Output");
if (useNewApi) {
try {
newRecordWriter.close(newApiTaskAttemptContext);
@@ -545,7 +549,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/983ceeee/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 cc2e234..5b0e62f 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
@@ -407,6 +407,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);
}
@@ -442,6 +453,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);
}
@@ -464,6 +486,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);
}
@@ -741,8 +774,8 @@ public class LogicalIOProcessorRuntimeTask extends RuntimeTask {
}
});
- eventRouterThread.setName("TezTaskEventRouter["
- + taskSpec.getTaskAttemptID().toString() + "]");
+ eventRouterThread.setName("TezTaskEventRouter{"
+ + taskSpec.getTaskAttemptID().toString() + "}");
eventRouterThread.start();
}
@@ -776,7 +809,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/983ceeee/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 3579e3f..6705020 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
@@ -282,7 +282,7 @@ public class TaskReporter implements TaskReporterInterface {
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/983ceeee/tez-runtime-library/src/main/java/org/apache/tez/http/HttpConnection.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/main/java/org/apache/tez/http/HttpConnection.java b/tez-runtime-library/src/main/java/org/apache/tez/http/HttpConnection.java
index 841e542..c94262a 100644
--- a/tez-runtime-library/src/main/java/org/apache/tez/http/HttpConnection.java
+++ b/tez-runtime-library/src/main/java/org/apache/tez/http/HttpConnection.java
@@ -266,7 +266,9 @@ public class HttpConnection extends BaseHttpConnection {
stopWatch.reset().start();
try {
if (input != null) {
- LOG.info("Closing input on " + logIdentifier);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Closing input on " + logIdentifier);
+ }
input.close();
input = null;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/983ceeee/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/983ceeee/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/983ceeee/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/983ceeee/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 8057be8..261f2e7 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
@@ -219,8 +219,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");
+ }
}
}
@@ -431,9 +433,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 = srcAttemptsRemaining.values().
toArray(new InputAttemptIdentifier[srcAttemptsRemaining.values().size()]);
@@ -443,7 +447,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, srcAttemptsRemaining.values()), null, false);
}
@@ -457,9 +463,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.iterator().next();
LOG.warn("Fetch Failure from host while connecting: " + host + ", attempt: " + firstAttempt
@@ -489,7 +497,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, srcAttemptsRemaining.values()), null, false);
}
// After this point, closing the stream and connection, should cause a
@@ -504,7 +514,10 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
while (!srcAttemptsRemaining.isEmpty() && failedInputs == null) {
if (isShutDown.get()) {
shutdownInternal(true);
- LOG.info("Fetcher already shutdown. Aborting queued fetches for " + srcAttemptsRemaining.size() + " inputs");
+ if (isDebugEnabled) {
+ LOG.debug("Fetcher already shutdown. Aborting queued fetches for " +
+ srcAttemptsRemaining.size() + " inputs");
+ }
return new HostFetchResult(new FetchResult(host, port, partition, srcAttemptsRemaining.values()), null,
false);
}
@@ -514,7 +527,10 @@ 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 " + srcAttemptsRemaining.size() + " inputs");
+ if (isDebugEnabled) {
+ LOG.debug("Fetcher already shutdown. Aborting reconnection and queued fetches for " +
+ srcAttemptsRemaining.size() + " inputs");
+ }
return new HostFetchResult(new FetchResult(host, port, partition, srcAttemptsRemaining.values()), null,
false);
}
@@ -527,8 +543,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, srcAttemptsRemaining.values()), failedInputs,
@@ -546,7 +564,10 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
Iterator<Entry<String, InputAttemptIdentifier>> iterator = srcAttemptsRemaining.entrySet().iterator();
while (iterator.hasNext()) {
if (isShutDown.get()) {
- LOG.info("Already shutdown. Skipping fetch for " + srcAttemptsRemaining.size() + " inputs");
+ if (isDebugEnabled) {
+ LOG.debug(
+ "Already shutdown. Skipping fetch for " + srcAttemptsRemaining.size() + " inputs");
+ }
break;
}
InputAttemptIdentifier srcAttemptId = iterator.next().getValue();
@@ -571,9 +592,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(),
@@ -582,9 +605,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;
}
if (failMissing) {
@@ -598,8 +624,11 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
InputAttemptIdentifier[] failedFetches = null;
if (failMissing && srcAttemptsRemaining.size() > 0) {
if (isShutDown.get()) {
- LOG.info("Already shutdown, not reporting fetch failures for: " + srcAttemptsRemaining.size() +
- " remaining inputs");
+ if (isDebugEnabled) {
+ LOG.debug(
+ "Already shutdown, not reporting fetch failures for: " + srcAttemptsRemaining.size() +
+ " remaining inputs");
+ }
} else {
failedFetches = srcAttemptsRemaining.values().
toArray(new InputAttemptIdentifier[srcAttemptsRemaining.values().size()]);
@@ -653,6 +682,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();
}
}
@@ -673,7 +705,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);
}
}
@@ -708,7 +740,9 @@ public class Fetcher extends CallableWithNdc<FetchResult> {
// Don't know which one was bad, so consider all of them as bad
return srcAttemptsRemaining.values().toArray(new InputAttemptIdentifier[srcAttemptsRemaining.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;
}
}
@@ -724,12 +758,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);
}
@@ -753,10 +789,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(),
@@ -765,7 +803,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 " +
@@ -795,8 +833,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/983ceeee/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/983ceeee/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 c7cc907..818cfaa 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
@@ -117,12 +117,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;
@@ -130,7 +134,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;
@@ -148,12 +152,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;
@@ -479,12 +487,28 @@ 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();
+ }
+
/**
* Build {@link org.apache.tez.http.HttpConnectionParams} from configuration
*
http://git-wip-us.apache.org/repos/asf/tez/blob/983ceeee/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/983ceeee/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 1977d5f..a7c1c59 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
@@ -226,11 +226,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);
@@ -268,7 +268,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 +", "
@@ -315,7 +315,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();
@@ -328,7 +328,7 @@ 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");
Thread.currentThread().interrupt();
break;
} else {
@@ -336,13 +336,13 @@ public class ShuffleManager implements FetcherCallback {
}
}
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");
break;
}
ListenableFuture<FetchResult> future = fetcherExecutor
@@ -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);
@@ -778,7 +783,7 @@ public class ShuffleManager implements FetcherCallback {
if (Thread.currentThread().isInterrupted()) {
//TODO: need to cleanup all FetchedInput (DiskFetchedInput, LocalDisFetchedInput), lockFile
//As of now relying on job cleanup (when all directories would be cleared)
- LOG.info("Thread interrupted. Need to cleanup the local dirs");
+ LOG.info(srcNameTrimmed + ": " + "Thread interrupted. Need to cleanup the local dirs");
}
if (!isShutdown.getAndSet(true)) {
// Shut down any pending fetchers
@@ -890,16 +895,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)");
+
+ }
}
@@ -907,15 +919,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");
}
}
@@ -944,7 +958,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()) {
@@ -965,9 +981,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/983ceeee/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/983ceeee/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 47df8f2..1b4031d 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
@@ -190,6 +190,9 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
public void shutDown() {
if (!stopped) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Fetcher stopped for host " + mapHost);
+ }
stopped = true;
// An interrupt will come in while shutting down the thread.
cleanupCurrentConnection(false);
@@ -264,14 +267,19 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
// 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, remaining.values())) {
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()};
@@ -282,8 +290,10 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
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) {
@@ -316,7 +326,9 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
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();
@@ -327,7 +339,9 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
Thread.currentThread().interrupt(); //reset status
}
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);
@@ -393,7 +407,9 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
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;
}
}
@@ -410,8 +426,10 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
// 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;
}
}
@@ -427,7 +445,9 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
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;
}
}
@@ -446,7 +466,9 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
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;
}
@@ -459,16 +481,19 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
}
// 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());
@@ -487,8 +512,10 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
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
@@ -654,7 +681,10 @@ class FetcherOrderedGrouped extends CallableWithNdc<Void> {
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/983ceeee/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 0a44a6b..eb2cece 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
@@ -17,27 +17,12 @@
*/
package org.apache.tez.runtime.library.common.shuffle.orderedgrouped;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.List;
-import java.util.Set;
-import java.util.TreeSet;
-
import com.google.common.annotations.VisibleForTesting;
-
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-
import com.google.common.base.Preconditions;
import org.apache.commons.io.FilenameUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.ChecksumFileSystem;
import org.apache.hadoop.fs.FileStatus;
@@ -61,12 +46,25 @@ 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;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
/**
@@ -229,10 +227,15 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
} 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(
@@ -262,10 +265,11 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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) {
@@ -320,8 +324,6 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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) {
@@ -329,7 +331,9 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
}
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;
@@ -384,9 +388,11 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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);
}
@@ -450,9 +456,9 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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();
@@ -474,7 +480,7 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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);
@@ -486,7 +492,7 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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());
}
@@ -631,7 +637,7 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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
@@ -648,7 +654,7 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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.");
@@ -715,7 +721,6 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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;
@@ -891,7 +896,7 @@ public class MergeManager implements FetchedInputAllocatorOrderedGrouped {
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 + "." +