You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by et...@apache.org on 2024/10/23 02:33:22 UTC
(celeborn) branch branch-tez-support updated: [CELEBORN-1540] Simplify Tez plugin implementation and improve integration testing
This is an automated email from the ASF dual-hosted git repository.
ethanfeng pushed a commit to branch branch-tez-support
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-tez-support by this push:
new 1937914b8 [CELEBORN-1540] Simplify Tez plugin implementation and improve integration testing
1937914b8 is described below
commit 1937914b8751fcd692f38f3e83ff15c57d01a66d
Author: 洪光威 <30...@users.noreply.github.com>
AuthorDate: Wed Oct 23 10:33:17 2024 +0800
[CELEBORN-1540] Simplify Tez plugin implementation and improve integration testing
Co-authored-by: hongguangwei <ho...@bytedance.com>
---
client-tez/tez-shaded/pom.xml | 42 +-
client-tez/tez/pom.xml | 5 +
.../celeborn/tez/plugin/util/CelebornTezUtils.java | 21 +-
.../apache/tez/dag/app/CelebornDagAppMaster.java | 32 +-
.../shuffle/impl/CelebornShuffleManager.java | 1058 +-------------
.../shuffle/orderedgrouped/CelebornScheduler.java | 1437 +-------------------
.../shuffle/orderedgrouped/CelebornShuffle.java | 391 +-----
.../CelebornTezShuffleDataFetcher.java | 15 +-
.../CelebornUnorderedPartitionedKVWriter.java | 91 +-
.../input/CelebornOrderedGroupedKVInput.java | 77 +-
.../library/input/CelebornUnorderedKVInput.java | 63 +-
.../output/CelebornOrderedPartitionedKVOutput.java | 207 +--
.../library/output/CelebornUnorderedKVOutput.java | 105 +-
.../CelebornUnorderedPartitionedKVOutput.java | 98 +-
.../runtime/library/output/OutputTestHelpers.java | 86 ++
.../output/TestOrderedPartitionedKVOutput.java | 124 ++
pom.xml | 17 -
.../tests/tez/TezCartesianProductTest.java | 80 --
.../tests/tez/TezCartesianProductTest.scala | 66 +
.../tests/tez/TezHashJoinBroadcastTest.java | 58 -
.../tests/tez/TezHashJoinBroadcastTest.scala | 38 +
.../apache/celeborn/tests/tez/TezHashJoinTest.java | 57 -
.../celeborn/tests/tez/TezHashJoinTest.scala | 37 +
.../tests/tez/TezIntegrationTestBase.scala | 297 ++++
...nTestBase.java => TezIntegrationTestBase2.java} | 188 +--
.../tests/tez/TezJoinIntegrationTestBase.java | 70 -
.../tests/tez/TezJoinIntegrationTestBase.scala | 72 +
.../tests/tez/TezOrderedWordCountTest.java | 86 --
.../tests/tez/TezOrderedWordCountTest.scala | 76 ++
.../celeborn/tests/tez/TezSortMergeJoinTest.java | 51 -
.../celeborn/tests/tez/TezSortMergeJoinTest.scala | 38 +
.../celeborn/tests/tez/TezWordCountTest.java | 71 -
.../celeborn/tests/tez/TezWordCountTest.scala | 61 +
tests/tez-it/src/test/resources/tez-site.xml | 6 -
34 files changed, 1313 insertions(+), 3908 deletions(-)
diff --git a/client-tez/tez-shaded/pom.xml b/client-tez/tez-shaded/pom.xml
index fd1489b0a..a249a34e7 100644
--- a/client-tez/tez-shaded/pom.xml
+++ b/client-tez/tez-shaded/pom.xml
@@ -38,25 +38,6 @@
<build>
<plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-assembly-plugin</artifactId>
- <configuration>
- <descriptors>
- <descriptor>src/main/resources/assembly.xml</descriptor>
- </descriptors>
- <appendAssemblyId>false</appendAssemblyId>
- </configuration>
- <executions>
- <execution>
- <id>make-assembly</id>
- <goals>
- <goal>single</goal>
- </goals>
- <phase>package</phase>
- </execution>
- </executions>
- </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
@@ -69,31 +50,22 @@
<relocation>
<pattern>com.google.common</pattern>
<shadedPattern>${shading.prefix}.com.google.common</shadedPattern>
+ <excludes>
+ <exclude>com.google.common.util.concurrent.*</exclude>
+ </excludes>
</relocation>
<relocation>
<pattern>io.netty</pattern>
<shadedPattern>${shading.prefix}.io.netty</shadedPattern>
</relocation>
<relocation>
- <pattern>org.scala-lang</pattern>
- <shadedPattern>${shading.prefix}.org.scala-lang</shadedPattern>
+ <pattern>org.apache.commons</pattern>
+ <shadedPattern>${shading.prefix}.org.apache.commons</shadedPattern>
</relocation>
<relocation>
<pattern>org.roaringbitmap</pattern>
<shadedPattern>${shading.prefix}.org.roaringbitmap</shadedPattern>
</relocation>
- <relocation>
- <pattern>org.apache.commons.lang3</pattern>
- <shadedPattern>${shading.prefix}.org.apache.commons.lang3</shadedPattern>
- </relocation>
- <relocation>
- <pattern>org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads</pattern>
- <shadedPattern>${shading.prefix}.org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads</shadedPattern>
- </relocation>
- <relocation>
- <pattern>org.apache.tez.common.TezCommonUtils</pattern>
- <shadedPattern>${shading.prefix}.org.apache.tez.common.TezCommonUtils</shadedPattern>
- </relocation>
</relocations>
<artifactSet>
<includes>
@@ -103,10 +75,10 @@
<include>com.google.guava:failureaccess</include>
<include>io.netty:*</include>
<include>org.apache.commons:commons-lang3</include>
+ <include>org.roaringbitmap:RoaringBitmap</include>
<include>org.scala-lang:scala-library</include>
+ <include>org.scala-lang:scala-reflect</include>
<include>org.lz4:lz4-java</include>
- <include>com.github.luben:zstd-jni</include>
- <include>org.roaringbitmap:RoaringBitmap</include>
</includes>
</artifactSet>
<filters>
diff --git a/client-tez/tez/pom.xml b/client-tez/tez/pom.xml
index a29306a83..1211303c7 100644
--- a/client-tez/tez/pom.xml
+++ b/client-tez/tez/pom.xml
@@ -188,6 +188,11 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/client-tez/tez/src/main/java/org/apache/celeborn/tez/plugin/util/CelebornTezUtils.java b/client-tez/tez/src/main/java/org/apache/celeborn/tez/plugin/util/CelebornTezUtils.java
index 1a77e6887..17cb1a62f 100644
--- a/client-tez/tez/src/main/java/org/apache/celeborn/tez/plugin/util/CelebornTezUtils.java
+++ b/client-tez/tez/src/main/java/org/apache/celeborn/tez/plugin/util/CelebornTezUtils.java
@@ -32,7 +32,6 @@ public class CelebornTezUtils {
public static final String TEZ_PREFIX = "tez.";
public static final String TEZ_CELEBORN_LM_HOST = "celeborn.lifecycleManager.host";
public static final String TEZ_CELEBORN_LM_PORT = "celeborn.lifecycleManager.port";
- public static final String TEZ_CELEBORN_USER = "celeborn.lifecycleManager.user";
public static final String TEZ_CELEBORN_APPLICATION_ID = "celeborn.applicationId";
public static final String TEZ_SHUFFLE_ID = "celeborn.tez.shuffle.id";
public static final String TEZ_BROADCAST_OR_ONETOONE = "celeborn.tez.broadcastOrOneToOne";
@@ -59,6 +58,26 @@ public class CelebornTezUtils {
}
}
+ public static Object getParentPrivateField(Object object, String name) {
+ try {
+ Field f = object.getClass().getSuperclass().getDeclaredField(name);
+ f.setAccessible(true);
+ return f.get(object);
+ } catch (Exception e) {
+ throw new CelebornRuntimeException(e.getMessage(), e);
+ }
+ }
+
+ public static void setParentPrivateField(Object object, String name, Object value) {
+ try {
+ Field f = object.getClass().getSuperclass().getDeclaredField(name);
+ f.setAccessible(true);
+ f.set(object, value);
+ } catch (Exception e) {
+ throw new CelebornRuntimeException(e.getMessage(), e);
+ }
+ }
+
public static String uniqueIdentifierToAttemptId(String uniqueIdentifier) {
if (uniqueIdentifier == null) {
throw new CelebornRuntimeException("uniqueIdentifier should not be null");
diff --git a/client-tez/tez/src/main/java/org/apache/tez/dag/app/CelebornDagAppMaster.java b/client-tez/tez/src/main/java/org/apache/tez/dag/app/CelebornDagAppMaster.java
index d495db631..866be9f35 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/dag/app/CelebornDagAppMaster.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/dag/app/CelebornDagAppMaster.java
@@ -30,9 +30,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import scala.Tuple2;
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.cli.GnuParser;
-import org.apache.commons.cli.Options;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
@@ -191,9 +188,6 @@ public class CelebornDagAppMaster extends DAGAppMaster {
edgeSourceConf.set(CelebornTezUtils.TEZ_CELEBORN_LM_HOST, lifecycleManagerHost);
edgeSourceConf.setInt(
CelebornTezUtils.TEZ_CELEBORN_LM_PORT, lifecycleManagerPort);
- edgeSourceConf.set(
- CelebornTezUtils.TEZ_CELEBORN_USER,
- lifecycleManager.getUserIdentifier().toString());
edgeSourceConf.set(
CelebornTezUtils.TEZ_BROADCAST_OR_ONETOONE,
String.valueOf(broadCastOrOneToOne));
@@ -333,9 +327,19 @@ public class CelebornDagAppMaster extends DAGAppMaster {
public static void main(String[] args) {
try {
- String s = "org.apache.tez.runtime.library.output.OrderedPartitionedKVOutput";
- if (s.equalsIgnoreCase(OrderedPartitionedKVOutput.class.getName())) {
- System.out.println("this is truth----");
+
+ boolean sessionModeCliOption = false;
+ for (int i = 0; i < args.length; i++) {
+ if (args[i].startsWith("-D")) {
+ String[] property = args[i].split("=");
+ if (property.length < 2) {
+ System.setProperty(property[0].substring(2), "");
+ } else {
+ System.setProperty(property[0].substring(2), property[1]);
+ }
+ } else if (args[i].contains("--session") || args[i].contains("-s")) {
+ sessionModeCliOption = true;
+ }
}
Thread.setDefaultUncaughtExceptionHandler(new YarnUncaughtExceptionHandler());
@@ -360,16 +364,6 @@ public class CelebornDagAppMaster extends DAGAppMaster {
String jobUserName = System.getenv(ApplicationConstants.Environment.USER.name());
- // Command line options
- Options opts = new Options();
- opts.addOption(
- TezConstants.TEZ_SESSION_MODE_CLI_OPTION,
- false,
- "Run Tez Application Master in Session mode");
-
- CommandLine cliParser = new GnuParser().parse(opts, args);
- boolean sessionModeCliOption = cliParser.hasOption(TezConstants.TEZ_SESSION_MODE_CLI_OPTION);
-
Logger.info(
"Creating CelebornDAGAppMaster for "
+ "applicationId={}"
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/CelebornShuffleManager.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/CelebornShuffleManager.java
index 173eff96d..1f2d80d8d 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/CelebornShuffleManager.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/impl/CelebornShuffleManager.java
@@ -14,79 +14,29 @@
*/
package org.apache.tez.runtime.library.common.shuffle.impl;
+import static org.apache.celeborn.tez.plugin.util.CelebornTezUtils.getParentPrivateField;
+
import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.nio.ByteBuffer;
-import java.text.DecimalFormat;
import java.util.*;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingDeque;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
-import javax.crypto.SecretKey;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Objects;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.ListeningExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.common.util.concurrent.*;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.LocalDirAllocator;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.util.Clock;
-import org.apache.hadoop.yarn.util.MonotonicClock;
-import org.apache.tez.common.CallableWithNdc;
-import org.apache.tez.common.GuavaShim;
-import org.apache.tez.common.Preconditions;
-import org.apache.tez.common.TezRuntimeFrameworkConfigs;
-import org.apache.tez.common.TezUtilsInternal;
-import org.apache.tez.common.counters.TaskCounter;
+import org.apache.tez.common.*;
import org.apache.tez.common.counters.TezCounter;
-import org.apache.tez.common.security.JobTokenSecretManager;
-import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.records.TezTaskAttemptID;
-import org.apache.tez.http.HttpConnectionParams;
-import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.InputContext;
import org.apache.tez.runtime.api.TaskFailureType;
-import org.apache.tez.runtime.api.events.InputReadErrorEvent;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.CompositeInputAttemptIdentifier;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
-import org.apache.tez.runtime.library.common.TezRuntimeUtils;
-import org.apache.tez.runtime.library.common.shuffle.FetchResult;
-import org.apache.tez.runtime.library.common.shuffle.FetchedInput;
-import org.apache.tez.runtime.library.common.shuffle.FetchedInput.Type;
-import org.apache.tez.runtime.library.common.shuffle.FetchedInputAllocator;
-import org.apache.tez.runtime.library.common.shuffle.Fetcher;
-import org.apache.tez.runtime.library.common.shuffle.Fetcher.FetcherBuilder;
-import org.apache.tez.runtime.library.common.shuffle.HostPort;
-import org.apache.tez.runtime.library.common.shuffle.InputAttemptFetchFailure;
-import org.apache.tez.runtime.library.common.shuffle.InputHost;
-import org.apache.tez.runtime.library.common.shuffle.InputHost.PartitionToInputs;
-import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
-import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils.FetchStatsLogger;
+import org.apache.tez.runtime.library.common.shuffle.*;
import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.CelebornTezReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,113 +44,36 @@ import org.slf4j.LoggerFactory;
import org.apache.celeborn.client.ShuffleClient;
import org.apache.celeborn.tez.plugin.util.CelebornTezUtils;
-// This only knows how to deal with a single srcIndex for a given targetIndex.
-// In case the src task generates multiple outputs for the same target Index
-// (multiple src-indices), modifications will be required.
public class CelebornShuffleManager extends ShuffleManager {
- private static final Logger LOG = LoggerFactory.getLogger(ShuffleManager.class);
- private static final Logger LOG_FETCH = LoggerFactory.getLogger(LOG.getName() + ".fetch");
- private static final FetchStatsLogger fetchStatsLogger = new FetchStatsLogger(LOG_FETCH, LOG);
+ private static final Logger LOG = LoggerFactory.getLogger(CelebornShuffleManager.class);
private final InputContext inputContext;
private final int numInputs;
+ private final ListeningExecutorService schedulerExecutor;
private final int shuffleId;
private final ApplicationAttemptId applicationAttemptId;
private final ShuffleClient shuffleClient;
-
- private final DecimalFormat mbpsFormat = new DecimalFormat("0.00");
-
private final FetchedInputAllocator inputManager;
- @VisibleForTesting final ListeningExecutorService fetcherExecutor;
-
- /** Executor for ReportCallable. */
- private ExecutorService reporterExecutor;
-
- /** Lock to sync failedEvents. */
- private final ReentrantLock reportLock = new ReentrantLock();
-
- /** Condition to wake up the thread notifying when events fail. */
- private final Condition reportCondition = reportLock.newCondition();
-
- /** Events reporting fetcher failed. */
- private final HashMap<InputReadErrorEvent, Integer> failedEvents = new HashMap<>();
-
- private final ListeningExecutorService schedulerExecutor;
- private final CelebornRunShuffleCallable schedulerCallable;
-
+ private final Set<CelebornFetcher> celebornRunningFetchers;
private final BlockingQueue<FetchedInput> completedInputs;
- private final AtomicBoolean inputReadyNotificationSent = new AtomicBoolean(false);
- @VisibleForTesting final BitSet completedInputSet;
- private final ConcurrentMap<HostPort, InputHost> knownSrcHosts;
private final BlockingQueue<InputHost> pendingHosts;
- private final Set<InputAttemptIdentifier> obsoletedInputs;
- private Set<CelebornFetcher> celebornRunningFetchers;
private final AtomicInteger numCompletedInputs = new AtomicInteger(0);
- private final AtomicInteger numFetchedSpills = new AtomicInteger(0);
-
- private final long startTime;
- private long lastProgressTime;
- private long totalBytesShuffledTillNow;
-
// Required to be held when manipulating pendingHosts
private final ReentrantLock lock = new ReentrantLock();
private final Condition wakeLoop = lock.newCondition();
private final int numFetchers;
- private final boolean asyncHttp;
-
- // Parameters required by Fetchers
- private final JobTokenSecretManager jobTokenSecretMgr;
- private final CompressionCodec codec;
- private final boolean localDiskFetchEnabled;
- private final boolean sharedFetchEnabled;
- private final boolean verifyDiskChecksum;
- private final boolean compositeFetch;
- private final boolean enableFetcherTestingErrors;
-
- private final int ifileBufferSize;
- private final boolean ifileReadAhead;
- private final int ifileReadAheadLength;
-
- /** Holds the time to wait for failures to batch them and send less events. */
private final int maxTimeToWaitForReportMillis;
-
private final String sourceDestNameTrimmed;
+ private final TezCounter shufflePhaseTime;
- private final int maxTaskOutputAtOnce;
-
- private final AtomicBoolean isShutdown = new AtomicBoolean(false);
-
- private long inputRecordsFromEvents;
- private long eventsReceived;
- private final TezCounter approximateInputRecords;
- private final TezCounter shuffledInputsCounter;
- private final TezCounter failedShufflesCounter;
- private final TezCounter bytesShuffledCounter;
- private final TezCounter decompressedDataSizeCounter;
- private final TezCounter bytesShuffledToDiskCounter;
- private final TezCounter bytesShuffledToMemCounter;
- private final TezCounter bytesShuffledDirectDiskCounter;
+ private final AtomicBoolean isShutdown;
+ private final long startTime;
private volatile Throwable shuffleError;
- private final HttpConnectionParams httpConnectionParams;
-
- private final LocalDirAllocator localDirAllocator;
- private final RawLocalFileSystem localFs;
- private final Path[] localDisks;
- private final String localhostName;
- private final int shufflePort;
-
- private final TezCounter shufflePhaseTime;
- private final TezCounter firstEventReceived;
- private final TezCounter lastEventReceived;
-
- // To track shuffleInfo events when finalMerge is disabled OR pipelined shuffle is enabled in
- // source.
- @VisibleForTesting final Map<Integer, ShuffleEventInfo> shuffleInfoEventsMap;
private final Set<Integer> successRssPartitionSet = new HashSet<>();
private final Set<Integer> runningRssPartitionMap = new HashSet<>();
private final Set<Integer> allRssPartition = Sets.newConcurrentHashSet();
@@ -210,8 +83,6 @@ public class CelebornShuffleManager extends ShuffleManager {
private final AtomicInteger numWithDataInput = new AtomicInteger(0);
private final boolean broadcastOrOneToOne;
- // TODO More counters - FetchErrors, speed?
-
public CelebornShuffleManager(
InputContext inputContext,
Configuration conf,
@@ -236,262 +107,40 @@ public class CelebornShuffleManager extends ShuffleManager {
codec,
inputAllocator);
this.inputContext = inputContext;
- this.numInputs = numInputs;
this.shuffleId = shuffleId;
this.applicationAttemptId = applicationAttemptId;
this.shuffleClient = shuffleClient;
this.broadcastOrOneToOne = broadcastOrOneToOne;
-
- this.approximateInputRecords =
- inputContext.getCounters().findCounter(TaskCounter.APPROXIMATE_INPUT_RECORDS);
- this.shuffledInputsCounter =
- inputContext.getCounters().findCounter(TaskCounter.NUM_SHUFFLED_INPUTS);
- this.failedShufflesCounter =
- inputContext.getCounters().findCounter(TaskCounter.NUM_FAILED_SHUFFLE_INPUTS);
- this.bytesShuffledCounter = inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES);
- this.decompressedDataSizeCounter =
- inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES_DECOMPRESSED);
- this.bytesShuffledToDiskCounter =
- inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES_TO_DISK);
- this.bytesShuffledToMemCounter =
- inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES_TO_MEM);
- this.bytesShuffledDirectDiskCounter =
- inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES_DISK_DIRECT);
-
- this.ifileBufferSize = bufferSize;
- this.ifileReadAhead = ifileReadAheadEnabled;
- this.ifileReadAheadLength = ifileReadAheadLength;
- this.codec = codec;
this.inputManager = inputAllocator;
- this.localDiskFetchEnabled =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH,
- TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH_DEFAULT);
- this.sharedFetchEnabled =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_SHARED_FETCH,
- TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_SHARED_FETCH_DEFAULT);
- this.verifyDiskChecksum =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_VERIFY_DISK_CHECKSUM,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_VERIFY_DISK_CHECKSUM_DEFAULT);
- this.maxTimeToWaitForReportMillis =
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_BATCH_WAIT,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_BATCH_WAIT_DEFAULT);
-
- this.shufflePhaseTime = inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_PHASE_TIME);
- this.firstEventReceived =
- inputContext.getCounters().findCounter(TaskCounter.FIRST_EVENT_RECEIVED);
- this.lastEventReceived =
- inputContext.getCounters().findCounter(TaskCounter.LAST_EVENT_RECEIVED);
- this.compositeFetch = ShuffleUtils.isTezShuffleHandler(conf);
-
- this.enableFetcherTestingErrors =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_ENABLE_TESTING_ERRORS,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_ENABLE_TESTING_ERRORS_DEFAULT);
-
- this.sourceDestNameTrimmed =
- TezUtilsInternal.cleanVertexName(inputContext.getSourceVertexName())
- + " -> "
- + TezUtilsInternal.cleanVertexName(inputContext.getTaskVertexName());
-
- completedInputSet = new BitSet(numInputs);
- /**
- * In case of pipelined shuffle, it is possible to get multiple FetchedInput per attempt. We do
- * not know upfront the number of spills from source.
- */
- completedInputs = new LinkedBlockingDeque<FetchedInput>();
- knownSrcHosts = new ConcurrentHashMap<HostPort, InputHost>();
- pendingHosts = new LinkedBlockingQueue<InputHost>();
- obsoletedInputs =
- Collections.newSetFromMap(new ConcurrentHashMap<InputAttemptIdentifier, Boolean>());
- celebornRunningFetchers =
+ this.celebornRunningFetchers =
Collections.newSetFromMap(new ConcurrentHashMap<CelebornFetcher, Boolean>());
-
- int maxConfiguredFetchers =
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES_DEFAULT);
-
- this.numFetchers = Math.min(maxConfiguredFetchers, numInputs);
-
- final ExecutorService fetcherRawExecutor;
- if (conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCHER_USE_SHARED_POOL,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCHER_USE_SHARED_POOL_DEFAULT)) {
- fetcherRawExecutor =
- inputContext.createTezFrameworkExecutorService(
- numFetchers, "Fetcher_B {" + sourceDestNameTrimmed + "} #%d");
- } else {
- fetcherRawExecutor =
- Executors.newFixedThreadPool(
- numFetchers,
- new ThreadFactoryBuilder()
- .setDaemon(true)
- .setNameFormat("Fetcher_B {" + sourceDestNameTrimmed + "} #%d")
- .build());
- }
- this.fetcherExecutor = MoreExecutors.listeningDecorator(fetcherRawExecutor);
-
- ExecutorService schedulerRawExecutor =
- Executors.newFixedThreadPool(
- 1,
- new ThreadFactoryBuilder()
- .setDaemon(true)
- .setNameFormat("ShuffleRunner {" + sourceDestNameTrimmed + "}")
- .build());
- this.schedulerExecutor = MoreExecutors.listeningDecorator(schedulerRawExecutor);
- this.schedulerCallable = new CelebornRunShuffleCallable(conf);
-
- this.startTime = System.currentTimeMillis();
- this.lastProgressTime = startTime;
-
- String auxiliaryService =
- conf.get(
- TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID,
- TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT);
- SecretKey shuffleSecret =
- ShuffleUtils.getJobTokenSecretFromTokenBytes(
- inputContext.getServiceConsumerMetaData(auxiliaryService));
- this.jobTokenSecretMgr = new JobTokenSecretManager(shuffleSecret);
- this.asyncHttp =
- conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP, false);
- httpConnectionParams = ShuffleUtils.getHttpConnectionParams(conf);
-
- this.localFs = (RawLocalFileSystem) FileSystem.getLocal(conf).getRaw();
-
- this.localDirAllocator = new LocalDirAllocator(TezRuntimeFrameworkConfigs.LOCAL_DIRS);
-
- this.localDisks =
- Iterables.toArray(localDirAllocator.getAllLocalPathsToRead(".", conf), Path.class);
- this.localhostName = inputContext.getExecutionContext().getHostName();
- final ByteBuffer shuffleMetaData = inputContext.getServiceProviderMetaData(auxiliaryService);
- this.shufflePort = ShuffleUtils.deserializeShuffleProviderMetaData(shuffleMetaData);
-
- /**
- * Setting to very high val can lead to Http 400 error. Cap it to 75; every attempt id would be
- * approximately 48 bytes; 48 * 75 = 3600 which should give some room for other info in URL.
- */
- this.maxTaskOutputAtOnce =
- Math.max(
- 1,
- Math.min(
- 75,
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE,
- TezRuntimeConfiguration
- .TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE_DEFAULT)));
-
- Arrays.sort(this.localDisks);
-
- shuffleInfoEventsMap = new ConcurrentHashMap<Integer, ShuffleEventInfo>();
-
- LOG.info(
- sourceDestNameTrimmed
- + ": numInputs="
- + numInputs
- + ", compressionCodec="
- + (codec == null ? "NoCompressionCodec" : codec.getClass().getName())
- + ", numFetchers="
- + numFetchers
- + ", ifileBufferSize="
- + ifileBufferSize
- + ", ifileReadAheadEnabled="
- + ifileReadAhead
- + ", ifileReadAheadLength="
- + ifileReadAheadLength
- + ", "
- + "localDiskFetchEnabled="
- + localDiskFetchEnabled
- + ", "
- + "sharedFetchEnabled="
- + sharedFetchEnabled
- + ", "
- + httpConnectionParams.toString()
- + ", maxTaskOutputAtOnce="
- + maxTaskOutputAtOnce
- + ", asyncHttp="
- + asyncHttp);
- }
-
- public void updateApproximateInputRecords(int delta) {
- if (delta <= 0) {
- return;
- }
- inputRecordsFromEvents += delta;
- eventsReceived++;
- approximateInputRecords.setValue((inputRecordsFromEvents / eventsReceived) * numInputs);
+ this.numInputs = numInputs;
+ this.numFetchers = (int) getParentPrivateField(this, "numFetchers");
+ this.schedulerExecutor =
+ (ListeningExecutorService) getParentPrivateField(this, "schedulerExecutor");
+ this.maxTimeToWaitForReportMillis =
+ (int) getParentPrivateField(this, "maxTimeToWaitForReportMillis");
+ this.isShutdown = (AtomicBoolean) getParentPrivateField(this, "isShutdown");
+ this.sourceDestNameTrimmed = (String) getParentPrivateField(this, "sourceDestNameTrimmed");
+ this.completedInputs =
+ (BlockingQueue<FetchedInput>) getParentPrivateField(this, "completedInputs");
+ this.pendingHosts = (BlockingQueue<InputHost>) getParentPrivateField(this, "pendingHosts");
+ this.shufflePhaseTime = (TezCounter) getParentPrivateField(this, "shufflePhaseTime");
+ this.startTime = (long) getParentPrivateField(this, "startTime");
}
+ @Override
public void run() throws IOException {
Preconditions.checkState(inputManager != null, "InputManager must be configured");
-
- if (maxTimeToWaitForReportMillis > 0) {
- reporterExecutor =
- Executors.newSingleThreadExecutor(
- new ThreadFactoryBuilder()
- .setDaemon(true)
- .setNameFormat("ShuffleRunner {" + sourceDestNameTrimmed + "}")
- .build());
- Future reporterFuture = reporterExecutor.submit(new ReporterCallable());
- }
-
- ListenableFuture<Void> runShuffleFuture = schedulerExecutor.submit(schedulerCallable);
+ // todo maxTimeToWaitForReportMillis
+ ListenableFuture<Void> runShuffleFuture =
+ schedulerExecutor.submit(new CelebornRunShuffleCallable());
Futures.addCallback(
runShuffleFuture, new SchedulerFutureCallback(), GuavaShim.directExecutor());
// Shutdown this executor once this task, and the callback complete.
schedulerExecutor.shutdown();
}
- private class ReporterCallable extends CallableWithNdc<Void> {
- /** Measures if the batching interval has ended. */
- private final Clock clock;
-
- ReporterCallable() {
- clock = new MonotonicClock();
- }
-
- @Override
- protected Void callInternal() throws Exception {
- long nextReport = 0;
- while (!isShutdown.get()) {
- try {
- reportLock.lock();
- while (failedEvents.isEmpty()) {
- boolean signaled =
- reportCondition.await(maxTimeToWaitForReportMillis, TimeUnit.MILLISECONDS);
- }
-
- long currentTime = clock.getTime();
- if (currentTime > nextReport) {
- if (failedEvents.size() > 0) {
- List<Event> failedEventsToSend = Lists.newArrayListWithCapacity(failedEvents.size());
- for (InputReadErrorEvent key : failedEvents.keySet()) {
- failedEventsToSend.add(
- InputReadErrorEvent.create(
- key.getDiagnostics(),
- key.getIndex(),
- key.getVersion(),
- failedEvents.get(key),
- key.isLocalFetch(),
- key.isDiskErrorAtSource(),
- localhostName));
- }
- inputContext.sendEvents(failedEventsToSend);
- failedEvents.clear();
- nextReport = currentTime + maxTimeToWaitForReportMillis;
- }
- }
- } finally {
- reportLock.unlock();
- }
- }
- return null;
- }
- }
-
private boolean isAllInputFetched() {
LOG.info(
"Check isAllInputFetched, numNoDataInput:{}, numWithDataInput:{},numInputs:{}, "
@@ -519,11 +168,7 @@ public class CelebornShuffleManager extends ShuffleManager {
private class CelebornRunShuffleCallable extends CallableWithNdc<Void> {
- private final Configuration conf;
-
- public CelebornRunShuffleCallable(Configuration conf) {
- this.conf = conf;
- }
+ public CelebornRunShuffleCallable() {}
@Override
protected Void callInternal() throws Exception {
@@ -626,147 +271,6 @@ public class CelebornShuffleManager extends ShuffleManager {
}
}
- private boolean validateInputAttemptForPipelinedShuffle(InputAttemptIdentifier input) {
- // For pipelined shuffle.
- // TODO: TEZ-2132 for error handling. As of now, fail fast if there is a different attempt
- if (input.canRetrieveInputInChunks()) {
- ShuffleEventInfo eventInfo = shuffleInfoEventsMap.get(input.getInputIdentifier());
- if (eventInfo != null && input.getAttemptNumber() != eventInfo.attemptNum) {
- if (eventInfo.scheduledForDownload || !eventInfo.eventsProcessed.isEmpty()) {
- IOException exception =
- new IOException(
- "Previous event already got scheduled for "
- + input
- + ". Previous attempt's data could have been already merged "
- + "to memory/disk outputs. Killing (self) this task early."
- + " currentAttemptNum="
- + eventInfo.attemptNum
- + ", eventsProcessed="
- + eventInfo.eventsProcessed
- + ", scheduledForDownload="
- + eventInfo.scheduledForDownload
- + ", newAttemptNum="
- + input.getAttemptNumber());
- String message = "Killing self as previous attempt data could have been consumed";
- killSelf(exception, message);
- return false;
- }
- }
- }
- return true;
- }
-
- void killSelf(Exception exception, String message) {
- LOG.error(message, exception);
- this.inputContext.killSelf(exception, message);
- }
-
- @VisibleForTesting
- Fetcher constructFetcherForHost(InputHost inputHost, Configuration conf) {
-
- Path lockDisk = null;
-
- if (sharedFetchEnabled) {
- // pick a single lock disk from the edge name's hashcode + host hashcode
- final int h = Math.abs(Objects.hashCode(this.sourceDestNameTrimmed, inputHost.getHost()));
- lockDisk = new Path(this.localDisks[h % this.localDisks.length], "locks");
- }
-
- FetcherBuilder fetcherBuilder =
- new FetcherBuilder(
- CelebornShuffleManager.this,
- httpConnectionParams,
- inputManager,
- inputContext,
- jobTokenSecretMgr,
- conf,
- localFs,
- localDirAllocator,
- lockDisk,
- localDiskFetchEnabled,
- sharedFetchEnabled,
- localhostName,
- shufflePort,
- asyncHttp,
- verifyDiskChecksum,
- compositeFetch,
- enableFetcherTestingErrors);
-
- if (codec != null) {
- fetcherBuilder.setCompressionParameters(codec);
- }
- fetcherBuilder.setIFileParams(ifileReadAhead, ifileReadAheadLength);
-
- // Remove obsolete inputs from the list being given to the fetcher. Also
- // remove from the obsolete list.
- PartitionToInputs pendingInputsOfOnePartitionRange = inputHost.clearAndGetOnePartitionRange();
- int includedMaps = 0;
- for (Iterator<InputAttemptIdentifier> inputIter =
- pendingInputsOfOnePartitionRange.getInputs().iterator();
- inputIter.hasNext(); ) {
- InputAttemptIdentifier input = inputIter.next();
-
- // For pipelined shuffle.
- if (!validateInputAttemptForPipelinedShuffle(input)) {
- continue;
- }
-
- // Avoid adding attempts which have already completed.
- boolean alreadyCompleted;
- if (input instanceof CompositeInputAttemptIdentifier) {
- CompositeInputAttemptIdentifier compositeInput = (CompositeInputAttemptIdentifier) input;
- int nextClearBit = completedInputSet.nextClearBit(compositeInput.getInputIdentifier());
- int maxClearBit =
- compositeInput.getInputIdentifier() + compositeInput.getInputIdentifierCount();
- alreadyCompleted = nextClearBit > maxClearBit;
- } else {
- alreadyCompleted = completedInputSet.get(input.getInputIdentifier());
- }
- // Avoid adding attempts which have already completed or have been marked as OBSOLETE
- if (alreadyCompleted || obsoletedInputs.contains(input)) {
- inputIter.remove();
- continue;
- }
-
- // Check if max threshold is met
- if (includedMaps >= maxTaskOutputAtOnce) {
- inputIter.remove();
- // add to inputHost
- inputHost.addKnownInput(
- pendingInputsOfOnePartitionRange.getPartition(),
- pendingInputsOfOnePartitionRange.getPartitionCount(),
- input);
- } else {
- includedMaps++;
- }
- }
- if (inputHost.getNumPendingPartitions() > 0) {
- pendingHosts.add(inputHost); // add it to queue
- }
- for (InputAttemptIdentifier input : pendingInputsOfOnePartitionRange.getInputs()) {
- ShuffleEventInfo eventInfo = shuffleInfoEventsMap.get(input.getInputIdentifier());
- if (eventInfo != null) {
- eventInfo.scheduledForDownload = true;
- }
- }
- fetcherBuilder.assignWork(
- inputHost.getHost(),
- inputHost.getPort(),
- pendingInputsOfOnePartitionRange.getPartition(),
- pendingInputsOfOnePartitionRange.getPartitionCount(),
- pendingInputsOfOnePartitionRange.getInputs());
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Created Fetcher for host: "
- + inputHost.getHost()
- + ", info: "
- + inputHost.getAdditionalInfo()
- + ", with inputs: "
- + pendingInputsOfOnePartitionRange);
- }
- return fetcherBuilder.build();
- }
-
CelebornFetcher constructFetcherForCeleborn(int partition) {
CelebornTezReader reader =
new CelebornTezReader(
@@ -783,39 +287,11 @@ public class CelebornShuffleManager extends ShuffleManager {
CompositeInputAttemptIdentifier srcAttemptIdentifier,
int srcPhysicalIndex) {
HostPort identifier = new HostPort(hostName, port);
- InputHost host = knownSrcHosts.get(identifier);
- if (host == null) {
- host = new InputHost(identifier);
- InputHost old = knownSrcHosts.putIfAbsent(identifier, host);
- if (old != null) {
- host = old;
- }
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- sourceDestNameTrimmed
- + ": "
- + "Adding input: "
- + srcAttemptIdentifier
- + ", to host: "
- + host);
- }
-
- if (!validateInputAttemptForPipelinedShuffle(srcAttemptIdentifier)) {
- return;
- }
- int inputIdentifier = srcAttemptIdentifier.getInputIdentifier();
- for (int i = 0; i < srcAttemptIdentifier.getInputIdentifierCount(); i++) {
- if (shuffleInfoEventsMap.get(inputIdentifier + i) == null) {
- shuffleInfoEventsMap.put(
- inputIdentifier + i, new ShuffleEventInfo(srcAttemptIdentifier.expand(i)));
- }
- }
-
- host.addKnownInput(
- srcPhysicalIndex, srcAttemptIdentifier.getInputIdentifierCount(), srcAttemptIdentifier);
+ super.addKnownInput(hostName, port, srcAttemptIdentifier, srcPhysicalIndex);
lock.lock();
try {
+ // pendingHosts is useless in celeborn
+ pendingHosts.remove(identifier);
if (broadcastOrOneToOne) {
String pathComponent = srcAttemptIdentifier.getPathComponent();
TezTaskAttemptID taskAttemptId =
@@ -854,375 +330,24 @@ public class CelebornShuffleManager extends ShuffleManager {
}
}
+ @Override
public void addCompletedInputWithNoData(InputAttemptIdentifier srcAttemptIdentifier) {
- int inputIdentifier = srcAttemptIdentifier.getInputIdentifier();
- LOG.debug("No input data exists for SrcTask: {}. Marking as complete.", inputIdentifier);
- lock.lock();
- try {
- if (!completedInputSet.get(inputIdentifier)) {
- NullFetchedInput fetchedInput = new NullFetchedInput(srcAttemptIdentifier);
- if (!srcAttemptIdentifier.canRetrieveInputInChunks()) {
- registerCompletedInput(fetchedInput);
- } else {
- registerCompletedInputForPipelinedShuffle(srcAttemptIdentifier, fetchedInput);
- }
- }
- // Awake the loop to check for termination.
- wakeLoop.signal();
- } finally {
- lock.unlock();
- }
+ super.addCompletedInputWithNoData(srcAttemptIdentifier);
numNoDataInput.incrementAndGet();
- }
-
- public void addCompletedInputWithData(
- InputAttemptIdentifier srcAttemptIdentifier, FetchedInput fetchedInput) throws IOException {
- // InputIdentifier inputIdentifier = srcAttemptIdentifier.getInputIdentifier();
- int inputIdentifier = srcAttemptIdentifier.getInputIdentifier();
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Received Data via Event: " + srcAttemptIdentifier + " to " + fetchedInput.getType());
- }
- // Count irrespective of whether this is a copy of an already fetched input
- lock.lock();
- try {
- lastProgressTime = System.currentTimeMillis();
- } finally {
- lock.unlock();
- }
-
- boolean committed = false;
- if (!completedInputSet.get(inputIdentifier)) {
- synchronized (completedInputSet) {
- if (!completedInputSet.get(inputIdentifier)) {
- fetchedInput.commit();
- committed = true;
- if (!srcAttemptIdentifier.canRetrieveInputInChunks()) {
- registerCompletedInput(fetchedInput);
- } else {
- registerCompletedInputForPipelinedShuffle(srcAttemptIdentifier, fetchedInput);
- }
- }
- }
- }
- if (!committed) {
- fetchedInput.abort(); // If this fails, the fetcher may attempt another
- // abort.
- } else {
- lock.lock();
- try {
- // Signal the wakeLoop to check for termination.
- wakeLoop.signal();
- } finally {
- lock.unlock();
- }
- }
- }
-
- protected synchronized void updateEventReceivedTime() {
- long relativeTime = System.currentTimeMillis() - startTime;
- if (firstEventReceived.getValue() == 0) {
- firstEventReceived.setValue(relativeTime);
- lastEventReceived.setValue(relativeTime);
- return;
- }
- lastEventReceived.setValue(relativeTime);
- }
-
- void obsoleteKnownInput(InputAttemptIdentifier srcAttemptIdentifier) {
- obsoletedInputs.add(srcAttemptIdentifier);
- // TODO NEWTEZ Maybe inform the fetcher about this. For now, this is used during the initial
- // fetch list construction.
- }
-
- /////////////////// End of Methods for InputEventHandler
- /////////////////// Methods from FetcherCallbackHandler
-
- /**
- * Placeholder for tracking shuffle events in case we get multiple spills info for the same
- * attempt.
- */
- static class ShuffleEventInfo {
- BitSet eventsProcessed;
- int finalEventId = -1; // 0 indexed
- int attemptNum;
- String id;
- boolean scheduledForDownload; // whether chunks got scheduled for download
-
- ShuffleEventInfo(InputAttemptIdentifier input) {
- this.id = input.getInputIdentifier() + "_" + input.getAttemptNumber();
- this.eventsProcessed = new BitSet();
- this.attemptNum = input.getAttemptNumber();
- }
-
- void spillProcessed(int spillId) {
- if (finalEventId != -1) {
- Preconditions.checkState(
- eventsProcessed.cardinality() <= (finalEventId + 1),
- "Wrong state. eventsProcessed cardinality="
- + eventsProcessed.cardinality()
- + " "
- + "finalEventId="
- + finalEventId
- + ", spillId="
- + spillId
- + ", "
- + toString());
- }
- eventsProcessed.set(spillId);
- }
-
- void setFinalEventId(int spillId) {
- finalEventId = spillId;
- }
-
- boolean isDone() {
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "finalEventId="
- + finalEventId
- + ", eventsProcessed cardinality="
- + eventsProcessed.cardinality());
- }
- return ((finalEventId != -1) && (finalEventId + 1) == eventsProcessed.cardinality());
- }
-
- public String toString() {
- return "[eventsProcessed="
- + eventsProcessed
- + ", finalEventId="
- + finalEventId
- + ", id="
- + id
- + ", attemptNum="
- + attemptNum
- + ", scheduledForDownload="
- + scheduledForDownload
- + "]";
- }
- }
-
- @Override
- public void fetchSucceeded(
- String host,
- InputAttemptIdentifier srcAttemptIdentifier,
- FetchedInput fetchedInput,
- long fetchedBytes,
- long decompressedLength,
- long copyDuration)
- throws IOException {
- int inputIdentifier = srcAttemptIdentifier.getInputIdentifier();
-
- // Count irrespective of whether this is a copy of an already fetched input
- lock.lock();
- try {
- lastProgressTime = System.currentTimeMillis();
- inputContext.notifyProgress();
- if (!completedInputSet.get(inputIdentifier)) {
- fetchedInput.commit();
- fetchStatsLogger.logIndividualFetchComplete(
- copyDuration,
- fetchedBytes,
- decompressedLength,
- fetchedInput.getType().toString(),
- srcAttemptIdentifier);
-
- // Processing counters for completed and commit fetches only. Need
- // additional counters for excessive fetches - which primarily comes
- // in after speculation or retries.
- shuffledInputsCounter.increment(1);
- bytesShuffledCounter.increment(fetchedBytes);
- if (fetchedInput.getType() == Type.MEMORY) {
- bytesShuffledToMemCounter.increment(fetchedBytes);
- } else if (fetchedInput.getType() == Type.DISK) {
- bytesShuffledToDiskCounter.increment(fetchedBytes);
- } else if (fetchedInput.getType() == Type.DISK_DIRECT) {
- bytesShuffledDirectDiskCounter.increment(fetchedBytes);
- }
- decompressedDataSizeCounter.increment(decompressedLength);
-
- if (!srcAttemptIdentifier.canRetrieveInputInChunks()) {
- registerCompletedInput(fetchedInput);
- } else {
- registerCompletedInputForPipelinedShuffle(srcAttemptIdentifier, fetchedInput);
- }
-
- totalBytesShuffledTillNow += fetchedBytes;
- logProgress();
- wakeLoop.signal();
- } else {
- fetchedInput.abort(); // If this fails, the fetcher may attempt another abort.
- }
- } finally {
- lock.unlock();
- }
- // TODO NEWTEZ Maybe inform fetchers, in case they have an alternate attempt of the same task in
- // their queue.
- }
-
- private void registerCompletedInput(FetchedInput fetchedInput) {
- lock.lock();
- try {
- maybeInformInputReady(fetchedInput);
- adjustCompletedInputs(fetchedInput);
- numFetchedSpills.getAndIncrement();
- } finally {
- lock.unlock();
- }
- }
-
- private void maybeInformInputReady(FetchedInput fetchedInput) {
- lock.lock();
- try {
- if (!(fetchedInput instanceof NullFetchedInput)) {
- completedInputs.add(fetchedInput);
- }
- if (!inputReadyNotificationSent.getAndSet(true)) {
- // TODO Should eventually be controlled by Inputs which are processing the data.
- inputContext.inputIsReady();
- }
- } finally {
- lock.unlock();
- }
- }
-
- private void adjustCompletedInputs(FetchedInput fetchedInput) {
- lock.lock();
- try {
- completedInputSet.set(fetchedInput.getInputAttemptIdentifier().getInputIdentifier());
-
- int numComplete = numCompletedInputs.incrementAndGet();
- if (numComplete == numInputs) {
- // Poison pill End of Input message to awake blocking take call
- if (fetchedInput instanceof NullFetchedInput) {
- completedInputs.add(fetchedInput);
- }
- LOG.info(
- "All inputs fetched for input vertex : " + inputContext.getInputOutputVertexNames());
- }
- } finally {
- lock.unlock();
- }
- }
-
- private void registerCompletedInputForPipelinedShuffle(
- InputAttemptIdentifier srcAttemptIdentifier, FetchedInput fetchedInput) {
- /**
- * For pipelinedshuffle it is possible to get multiple spills. Claim success only when all
- * spills pertaining to an attempt are done.
- */
- if (!validateInputAttemptForPipelinedShuffle(srcAttemptIdentifier)) {
- return;
- }
-
- int inputIdentifier = srcAttemptIdentifier.getInputIdentifier();
- ShuffleEventInfo eventInfo = shuffleInfoEventsMap.get(inputIdentifier);
-
- // for empty partition case
- if (eventInfo == null && fetchedInput instanceof NullFetchedInput) {
- eventInfo = new ShuffleEventInfo(srcAttemptIdentifier);
- shuffleInfoEventsMap.put(inputIdentifier, eventInfo);
- }
-
- assert (eventInfo != null);
- eventInfo.spillProcessed(srcAttemptIdentifier.getSpillEventId());
- numFetchedSpills.getAndIncrement();
-
- if (srcAttemptIdentifier.getFetchTypeInfo() == InputAttemptIdentifier.SPILL_INFO.FINAL_UPDATE) {
- eventInfo.setFinalEventId(srcAttemptIdentifier.getSpillEventId());
- }
-
- lock.lock();
- try {
- /**
- * When fetch is complete for a spill, add it to completedInputs to ensure that it is
- * available for downstream processing. Final success will be claimed only when all spills are
- * downloaded from the source.
- */
- maybeInformInputReady(fetchedInput);
-
- // check if we downloaded all spills pertaining to this InputAttemptIdentifier
- if (eventInfo.isDone()) {
- adjustCompletedInputs(fetchedInput);
- shuffleInfoEventsMap.remove(srcAttemptIdentifier.getInputIdentifier());
- }
- } finally {
- lock.unlock();
- }
-
- if (LOG.isTraceEnabled()) {
- LOG.trace("eventInfo " + eventInfo.toString());
- }
- }
-
- private void reportFatalError(Throwable exception, String message) {
- LOG.error(message);
- inputContext.reportFailure(TaskFailureType.NON_FATAL, exception, message);
- }
-
- @Override
- public void fetchFailed(
- String host, InputAttemptFetchFailure inputAttemptFetchFailure, boolean connectFailed) {
- // TODO NEWTEZ. Implement logic to report fetch failures after a threshold.
- // For now, reporting immediately.
- InputAttemptIdentifier srcAttemptIdentifier =
- inputAttemptFetchFailure.getInputAttemptIdentifier();
LOG.info(
- "{}: Fetch failed for src: {} InputIdentifier: {}, connectFailed: {}, "
- + "local fetch: {}, remote fetch failure reported as local failure: {})",
- sourceDestNameTrimmed,
- srcAttemptIdentifier,
- srcAttemptIdentifier,
- connectFailed,
- inputAttemptFetchFailure.isLocalFetch(),
- inputAttemptFetchFailure.isDiskErrorAtSource());
- failedShufflesCounter.increment(1);
- inputContext.notifyProgress();
- if (srcAttemptIdentifier == null) {
- reportFatalError(null, "Received fetchFailure for an unknown src (null)");
- } else {
- InputReadErrorEvent readError =
- InputReadErrorEvent.create(
- "Fetch failure while fetching from "
- + TezRuntimeUtils.getTaskAttemptIdentifier(
- inputContext.getSourceVertexName(),
- srcAttemptIdentifier.getInputIdentifier(),
- srcAttemptIdentifier.getAttemptNumber()),
- srcAttemptIdentifier.getInputIdentifier(),
- srcAttemptIdentifier.getAttemptNumber(),
- inputAttemptFetchFailure.isLocalFetch(),
- inputAttemptFetchFailure.isDiskErrorAtSource(),
- localhostName);
- if (maxTimeToWaitForReportMillis > 0) {
- try {
- reportLock.lock();
- failedEvents.merge(readError, 1, (a, b) -> a + b);
- reportCondition.signal();
- } finally {
- reportLock.unlock();
- }
- } else {
- List<Event> events = Lists.newArrayListWithCapacity(1);
- events.add(readError);
- inputContext.sendEvents(events);
- }
- }
+ "AddCompletedInputWithNoData, numNoDataInput:{}, numWithDataInput:{},numInputs:{}, "
+ + "successRssPartitionSet:{}, allRssPartition:{}.",
+ numNoDataInput,
+ numWithDataInput,
+ numInputs,
+ successRssPartitionSet,
+ allRssPartition);
}
- /////////////////// End of Methods from FetcherCallbackHandler
public void shutdown() throws InterruptedException {
- 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(sourceDestNameTrimmed + ": " + "Thread interrupted. Need to cleanup the local dirs");
- }
- if (!isShutdown.getAndSet(true)) {
- // Shut down any pending fetchers
- LOG.info(
- "Shutting down pending fetchers on source"
- + sourceDestNameTrimmed
- + ": "
- + celebornRunningFetchers.size());
+
+ if (!isShutdown.get()) {
+ super.shutdown();
lock.lock();
try {
wakeLoop.signal(); // signal the fetch-scheduler
@@ -1238,16 +363,6 @@ public class CelebornShuffleManager extends ShuffleManager {
} finally {
lock.unlock();
}
-
- if (this.schedulerExecutor != null && !this.schedulerExecutor.isShutdown()) {
- this.schedulerExecutor.shutdownNow();
- }
- if (this.reporterExecutor != null && !this.reporterExecutor.isShutdown()) {
- this.reporterExecutor.shutdownNow();
- }
- if (this.fetcherExecutor != null && !this.fetcherExecutor.isShutdown()) {
- this.fetcherExecutor.shutdownNow(); // Interrupts all running fetchers.
- }
}
}
@@ -1284,6 +399,7 @@ public class CelebornShuffleManager extends ShuffleManager {
break;
}
LOG.info("GetNextInput, out loop");
+ Thread.sleep(1000);
}
} else {
fetchedInput = completedInputs.take();
@@ -1297,86 +413,6 @@ public class CelebornShuffleManager extends ShuffleManager {
return fetchedInput;
}
- public int getNumInputs() {
- return numInputs;
- }
-
- public float getNumCompletedInputsFloat() {
- return numCompletedInputs.floatValue();
- }
-
- /////////////////// End of methods for walking the available inputs
-
- /**
- * Fake input that is added to the completed input list in case an input does not have any data.
- */
- @VisibleForTesting
- static class NullFetchedInput extends FetchedInput {
-
- public NullFetchedInput(InputAttemptIdentifier inputAttemptIdentifier) {
- super(inputAttemptIdentifier, null);
- }
-
- @Override
- public Type getType() {
- return Type.MEMORY;
- }
-
- @Override
- public long getSize() {
- return -1;
- }
-
- @Override
- public OutputStream getOutputStream() throws IOException {
- throw new UnsupportedOperationException("Not supported for NullFetchedInput");
- }
-
- @Override
- public InputStream getInputStream() throws IOException {
- throw new UnsupportedOperationException("Not supported for NullFetchedInput");
- }
-
- @Override
- public void commit() throws IOException {
- throw new UnsupportedOperationException("Not supported for NullFetchedInput");
- }
-
- @Override
- public void abort() throws IOException {
- throw new UnsupportedOperationException("Not supported for NullFetchedInput");
- }
-
- @Override
- public void free() {
- throw new UnsupportedOperationException("Not supported for NullFetchedInput");
- }
- }
-
- private final AtomicInteger nextProgressLineEventCount = new AtomicInteger(0);
-
- private void logProgress() {
- int inputsDone = numCompletedInputs.get();
-
- 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)");
- }
- }
-
private class SchedulerFutureCallback implements FutureCallback<Void> {
@Override
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornScheduler.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornScheduler.java
index 536793c63..bd64eac79 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornScheduler.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornScheduler.java
@@ -14,190 +14,56 @@
*/
package org.apache.tez.runtime.library.common.shuffle.orderedgrouped;
+import static org.apache.celeborn.tez.plugin.util.CelebornTezUtils.getParentPrivateField;
+
import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.text.DecimalFormat;
-import java.util.ArrayList;
-import java.util.BitSet;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
-import java.util.Set;
+import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.DelayQueue;
-import java.util.concurrent.Delayed;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-import javax.crypto.SecretKey;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.LinkedListMultimap;
-import com.google.common.collect.ListMultimap;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.RawLocalFileSystem;
-import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.tez.common.CallableWithNdc;
-import org.apache.tez.common.Preconditions;
-import org.apache.tez.common.TezUtilsInternal;
-import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.common.counters.TezCounter;
-import org.apache.tez.common.security.JobTokenSecretManager;
-import org.apache.tez.dag.api.TezConfiguration;
-import org.apache.tez.http.HttpConnectionParams;
-import org.apache.tez.runtime.api.Event;
+import org.apache.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.dag.records.TezTaskID;
import org.apache.tez.runtime.api.InputContext;
-import org.apache.tez.runtime.api.events.InputReadErrorEvent;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.CompositeInputAttemptIdentifier;
import org.apache.tez.runtime.library.common.InputAttemptIdentifier;
-import org.apache.tez.runtime.library.common.TezRuntimeUtils;
-import org.apache.tez.runtime.library.common.shuffle.HostPort;
-import org.apache.tez.runtime.library.common.shuffle.InputAttemptFetchFailure;
-import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
-import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils.FetchStatsLogger;
-import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.MapHost.HostPortPartition;
-import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.MapOutput.Type;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.celeborn.client.ShuffleClient;
class CelebornScheduler extends ShuffleScheduler {
- @VisibleForTesting static final String SHUFFLE_ERR_GRP_NAME = "Shuffle Errors";
-
- private final AtomicLong shuffleStart = new AtomicLong(0);
-
private static final Logger LOG = LoggerFactory.getLogger(ShuffleScheduler.class);
- private static final Logger LOG_FETCH = LoggerFactory.getLogger(LOG.getName() + ".fetch");
- private static final FetchStatsLogger fetchStatsLogger = new FetchStatsLogger(LOG_FETCH, LOG);
-
- static final long INITIAL_PENALTY = 2000L; // 2 seconds
- private static final float PENALTY_GROWTH_RATE = 1.3f;
-
- private final BitSet finishedMaps;
- private final int numInputs;
- private int numFetchedSpills;
-
- @VisibleForTesting
- final Map<HostPortPartition, MapHost> mapLocations = new HashMap<HostPortPartition, MapHost>();
- // TODO Clean this and other maps at some point
- @VisibleForTesting
- final ConcurrentMap<PathPartition, InputAttemptIdentifier> pathToIdentifierMap =
- new ConcurrentHashMap<PathPartition, InputAttemptIdentifier>();
-
- // To track shuffleInfo events when finalMerge is disabled in source or pipelined shuffle is
- // enabled in source.
- @VisibleForTesting final Map<Integer, ShuffleEventInfo> pipelinedShuffleInfoEventsMap;
-
- @VisibleForTesting final Set<MapHost> pendingHosts = new HashSet<MapHost>();
- private final Set<InputAttemptIdentifier> obsoleteInputs = new HashSet<InputAttemptIdentifier>();
- private final AtomicBoolean isShutdown = new AtomicBoolean(false);
+ private final AtomicBoolean isShutdown;
private final Random random = new Random(System.currentTimeMillis());
- private final DelayQueue<Penalty> penalties = new DelayQueue<Penalty>();
- private final Referee referee;
-
- @VisibleForTesting
- final Map<InputAttemptIdentifier, IntWritable> failureCounts =
- new HashMap<InputAttemptIdentifier, IntWritable>();
-
- final Set<HostPort> uniqueHosts = Sets.newHashSet();
- private final Map<HostPort, IntWritable> hostFailures = new HashMap<HostPort, IntWritable>();
- private final InputContext inputContext;
- private final TezCounter shuffledInputsCounter;
- private final TezCounter skippedInputCounter;
- private final TezCounter reduceShuffleBytes;
- private final TezCounter reduceBytesDecompressed;
- @VisibleForTesting final TezCounter failedShuffleCounter;
- private final TezCounter bytesShuffledToDisk;
- private final TezCounter bytesShuffledToDiskDirect;
- private final TezCounter bytesShuffledToMem;
- private final TezCounter firstEventReceived;
- private final TezCounter lastEventReceived;
private final String srcNameTrimmed;
- @VisibleForTesting final AtomicInteger remainingMaps;
-
- private final long startTime;
- @VisibleForTesting long lastProgressTime;
- @VisibleForTesting long failedShufflesSinceLastCompletion;
private final int numFetchers;
+ private final ListeningExecutorService fetcherExecutor;
private final Set<CelebornTezShuffleDataFetcher> celebornRunningFetchers =
Collections.newSetFromMap(new ConcurrentHashMap<CelebornTezShuffleDataFetcher, Boolean>());
-
- private final ListeningExecutorService fetcherExecutor;
-
- private final HttpConnectionParams httpConnectionParams;
- private final FetchedInputAllocatorOrderedGrouped allocator;
- private final ExceptionReporter exceptionReporter;
+ private volatile Thread shuffleSchedulerThread = null;
private final MergeManager mergeManager;
- private final JobTokenSecretManager jobTokenSecretManager;
- private final boolean ifileReadAhead;
- private final int ifileReadAheadLength;
- private final CompressionCodec codec;
- private final Configuration conf;
- private final RawLocalFileSystem localFs;
- private final boolean localDiskFetchEnabled;
- private final String localHostname;
- private final int shufflePort;
- private final boolean asyncHttp;
- private final boolean sslShuffle;
+ private final InputContext inputContext;
+ private final ExceptionReporter exceptionReporter;
+ private final int numInputs;
+
+ // celeborn
private final int shuffleId;
private final ApplicationAttemptId applicationAttemptId;
private final ShuffleClient shuffleClient;
-
- private final TezCounter ioErrsCounter;
- private final TezCounter wrongLengthErrsCounter;
- private final TezCounter badIdErrsCounter;
- private final TezCounter wrongMapErrsCounter;
- private final TezCounter connectionErrsCounter;
- private final TezCounter wrongReduceErrsCounter;
-
- private final int maxTaskOutputAtOnce;
- private final int maxFetchFailuresBeforeReporting;
- private final boolean reportReadErrorImmediately;
- private final int maxFailedUniqueFetches;
- private final int abortFailureLimit;
-
- private final int minFailurePerHost;
- private final float hostFailureFraction;
- private final float maxStallTimeFraction;
- private final float minReqProgressFraction;
- private final float maxAllowedFailedFetchFraction;
- private final boolean checkFailedFetchSinceLastCompletion;
- private final boolean verifyDiskChecksum;
- private final boolean compositeFetch;
- private final boolean enableFetcherTestingErrors;
- private volatile Thread shuffleSchedulerThread = null;
- private final int maxPenaltyTime;
-
- private long totalBytesShuffledTillNow = 0;
- private final DecimalFormat mbpsFormat = new DecimalFormat("0.00");
-
- // celeborn
private final Map<Integer, MapHost> runningRssPartitionMap = new HashMap<>();
private final Set<Integer> successRssPartitionSet = Sets.newConcurrentHashSet();
@@ -205,6 +71,8 @@ class CelebornScheduler extends ShuffleScheduler {
private final Map<Integer, Set<InputAttemptIdentifier>> partitionIdToSuccessMapTaskAttempts =
new HashMap<>();
+ final Map<Integer, Set<TezTaskID>> partitionIdToSuccessTezTasks = new HashMap<>();
+ private final TezCounter skippedInputCounter;
public CelebornScheduler(
InputContext inputContext,
@@ -235,260 +103,18 @@ class CelebornScheduler extends ShuffleScheduler {
ifileReadAheadLength,
srcNameTrimmed);
this.inputContext = inputContext;
- this.conf = conf;
- this.localFs = (RawLocalFileSystem) FileSystem.getLocal(conf).getRaw();
this.exceptionReporter = exceptionReporter;
- this.allocator = allocator;
- this.mergeManager = mergeManager;
- this.numInputs = numberOfInputs;
- int abortFailureLimitConf =
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_SOURCE_ATTEMPT_ABORT_LIMIT,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_SOURCE_ATTEMPT_ABORT_LIMIT_DEFAULT);
- if (abortFailureLimitConf <= -1) {
- abortFailureLimit = Math.max(15, numberOfInputs / 10);
- } else {
- // No upper cap, as user is setting this intentionally
- abortFailureLimit = abortFailureLimitConf;
- }
- remainingMaps = new AtomicInteger(numberOfInputs);
- finishedMaps = new BitSet(numberOfInputs);
- this.ifileReadAhead = ifileReadAhead;
- this.ifileReadAheadLength = ifileReadAheadLength;
this.srcNameTrimmed = srcNameTrimmed;
this.shuffleClient = shuffleClient;
this.shuffleId = shuffleId;
this.applicationAttemptId = applicationAttemptId;
- this.codec = codec;
- int configuredNumFetchers =
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES_DEFAULT);
- numFetchers = Math.min(configuredNumFetchers, numInputs);
-
- localDiskFetchEnabled =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH,
- TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH_DEFAULT);
-
- this.minFailurePerHost =
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MIN_FAILURES_PER_HOST,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MIN_FAILURES_PER_HOST_DEFAULT);
- Preconditions.checkArgument(
- minFailurePerHost >= 0,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MIN_FAILURES_PER_HOST
- + "="
- + minFailurePerHost
- + " should not be negative");
-
- this.hostFailureFraction =
- conf.getFloat(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ACCEPTABLE_HOST_FETCH_FAILURE_FRACTION,
- TezRuntimeConfiguration
- .TEZ_RUNTIME_SHUFFLE_ACCEPTABLE_HOST_FETCH_FAILURE_FRACTION_DEFAULT);
-
- this.maxStallTimeFraction =
- conf.getFloat(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MAX_STALL_TIME_FRACTION,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MAX_STALL_TIME_FRACTION_DEFAULT);
- Preconditions.checkArgument(
- maxStallTimeFraction >= 0,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MAX_STALL_TIME_FRACTION
- + "="
- + maxStallTimeFraction
- + " should not be negative");
-
- this.minReqProgressFraction =
- conf.getFloat(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MIN_REQUIRED_PROGRESS_FRACTION,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MIN_REQUIRED_PROGRESS_FRACTION_DEFAULT);
- Preconditions.checkArgument(
- minReqProgressFraction >= 0,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MIN_REQUIRED_PROGRESS_FRACTION
- + "="
- + minReqProgressFraction
- + " should not be negative");
-
- this.maxAllowedFailedFetchFraction =
- conf.getFloat(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MAX_ALLOWED_FAILED_FETCH_ATTEMPT_FRACTION,
- TezRuntimeConfiguration
- .TEZ_RUNTIME_SHUFFLE_MAX_ALLOWED_FAILED_FETCH_ATTEMPT_FRACTION_DEFAULT);
- Preconditions.checkArgument(
- maxAllowedFailedFetchFraction >= 0,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MAX_ALLOWED_FAILED_FETCH_ATTEMPT_FRACTION
- + "="
- + maxAllowedFailedFetchFraction
- + " should not be negative");
-
- this.checkFailedFetchSinceLastCompletion =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FAILED_CHECK_SINCE_LAST_COMPLETION,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FAILED_CHECK_SINCE_LAST_COMPLETION_DEFAULT);
-
- this.localHostname = inputContext.getExecutionContext().getHostName();
- String auxiliaryService =
- conf.get(
- TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID,
- TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT);
- final ByteBuffer shuffleMetadata = inputContext.getServiceProviderMetaData(auxiliaryService);
- this.shufflePort = ShuffleUtils.deserializeShuffleProviderMetaData(shuffleMetadata);
-
- this.referee = new Referee();
- // Counters used by the ShuffleScheduler
- this.shuffledInputsCounter =
- inputContext.getCounters().findCounter(TaskCounter.NUM_SHUFFLED_INPUTS);
- this.reduceShuffleBytes = inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES);
- this.reduceBytesDecompressed =
- inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES_DECOMPRESSED);
- this.failedShuffleCounter =
- inputContext.getCounters().findCounter(TaskCounter.NUM_FAILED_SHUFFLE_INPUTS);
- this.bytesShuffledToDisk =
- inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES_TO_DISK);
- this.bytesShuffledToDiskDirect =
- inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES_DISK_DIRECT);
- this.bytesShuffledToMem =
- inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_BYTES_TO_MEM);
-
- // Counters used by Fetchers
- ioErrsCounter =
- inputContext
- .getCounters()
- .findCounter(SHUFFLE_ERR_GRP_NAME, ShuffleErrors.IO_ERROR.toString());
- wrongLengthErrsCounter =
- inputContext
- .getCounters()
- .findCounter(SHUFFLE_ERR_GRP_NAME, ShuffleErrors.WRONG_LENGTH.toString());
- badIdErrsCounter =
- inputContext
- .getCounters()
- .findCounter(SHUFFLE_ERR_GRP_NAME, ShuffleErrors.BAD_ID.toString());
- wrongMapErrsCounter =
- inputContext
- .getCounters()
- .findCounter(SHUFFLE_ERR_GRP_NAME, ShuffleErrors.WRONG_MAP.toString());
- connectionErrsCounter =
- inputContext
- .getCounters()
- .findCounter(SHUFFLE_ERR_GRP_NAME, ShuffleErrors.CONNECTION.toString());
- wrongReduceErrsCounter =
- inputContext
- .getCounters()
- .findCounter(SHUFFLE_ERR_GRP_NAME, ShuffleErrors.WRONG_REDUCE.toString());
-
- this.startTime = startTime;
- this.lastProgressTime = startTime;
-
- this.sslShuffle =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL_DEFAULT);
- this.asyncHttp =
- conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP, false);
- this.httpConnectionParams = ShuffleUtils.getHttpConnectionParams(conf);
- SecretKey jobTokenSecret =
- ShuffleUtils.getJobTokenSecretFromTokenBytes(
- inputContext.getServiceConsumerMetaData(auxiliaryService));
- this.jobTokenSecretManager = new JobTokenSecretManager(jobTokenSecret);
-
- final ExecutorService fetcherRawExecutor;
- if (conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCHER_USE_SHARED_POOL,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCHER_USE_SHARED_POOL_DEFAULT)) {
- fetcherRawExecutor =
- inputContext.createTezFrameworkExecutorService(
- numFetchers, "Fetcher_O {" + srcNameTrimmed + "} #%d");
- } else {
- fetcherRawExecutor =
- Executors.newFixedThreadPool(
- numFetchers,
- new ThreadFactoryBuilder()
- .setDaemon(true)
- .setNameFormat("Fetcher_O {" + srcNameTrimmed + "} #%d")
- .build());
- }
- this.fetcherExecutor = MoreExecutors.listeningDecorator(fetcherRawExecutor);
-
- this.maxFailedUniqueFetches = Math.min(numberOfInputs, 5);
- referee.start();
- this.maxFetchFailuresBeforeReporting =
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT_DEFAULT);
- this.reportReadErrorImmediately =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR_DEFAULT);
- this.verifyDiskChecksum =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_VERIFY_DISK_CHECKSUM,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_VERIFY_DISK_CHECKSUM_DEFAULT);
-
- /**
- * Setting to very high val can lead to Http 400 error. Cap it to 75; every attempt id would be
- * approximately 48 bytes; 48 * 75 = 3600 which should give some room for other info in URL.
- */
- this.maxTaskOutputAtOnce =
- Math.max(
- 1,
- Math.min(
- 75,
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE,
- TezRuntimeConfiguration
- .TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE_DEFAULT)));
-
- this.skippedInputCounter =
- inputContext.getCounters().findCounter(TaskCounter.NUM_SKIPPED_INPUTS);
- this.firstEventReceived =
- inputContext.getCounters().findCounter(TaskCounter.FIRST_EVENT_RECEIVED);
- this.lastEventReceived =
- inputContext.getCounters().findCounter(TaskCounter.LAST_EVENT_RECEIVED);
- this.compositeFetch = ShuffleUtils.isTezShuffleHandler(conf);
- this.maxPenaltyTime =
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_HOST_PENALTY_TIME_LIMIT_MS,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_HOST_PENALTY_TIME_LIMIT_MS_DEFAULT);
-
- this.enableFetcherTestingErrors =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_ENABLE_TESTING_ERRORS,
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_ENABLE_TESTING_ERRORS_DEFAULT);
-
- pipelinedShuffleInfoEventsMap = Maps.newConcurrentMap();
- LOG.info(
- "ShuffleScheduler running for sourceVertex: "
- + inputContext.getSourceVertexName()
- + " with configuration: "
- + "maxFetchFailuresBeforeReporting="
- + maxFetchFailuresBeforeReporting
- + ", reportReadErrorImmediately="
- + reportReadErrorImmediately
- + ", maxFailedUniqueFetches="
- + maxFailedUniqueFetches
- + ", abortFailureLimit="
- + abortFailureLimit
- + ", maxTaskOutputAtOnce="
- + maxTaskOutputAtOnce
- + ", numFetchers="
- + numFetchers
- + ", hostFailureFraction="
- + hostFailureFraction
- + ", minFailurePerHost="
- + minFailurePerHost
- + ", maxAllowedFailedFetchFraction="
- + maxAllowedFailedFetchFraction
- + ", maxStallTimeFraction="
- + maxStallTimeFraction
- + ", minReqProgressFraction="
- + minReqProgressFraction
- + ", checkFailedFetchSinceLastCompletion="
- + checkFailedFetchSinceLastCompletion
- + ", asyncHttp="
- + asyncHttp
- + ", enableFetcherTestingErrors="
- + enableFetcherTestingErrors);
+ this.mergeManager = mergeManager;
+ this.numInputs = numberOfInputs;
+ this.numFetchers = (int) getParentPrivateField(this, "numFetchers");
+ this.fetcherExecutor =
+ (ListeningExecutorService) getParentPrivateField(this, "fetcherExecutor");
+ this.isShutdown = (AtomicBoolean) getParentPrivateField(this, "isShutdown");
+ this.skippedInputCounter = (TezCounter) getParentPrivateField(this, "skippedInputCounter");
}
public void start() throws Exception {
@@ -498,762 +124,30 @@ class CelebornScheduler extends ShuffleScheduler {
schedulerCallable.call();
}
- public void close() {
- try {
- if (!isShutdown.getAndSet(true)) {
- try {
- logProgress();
- } catch (Exception e) {
- LOG.warn(
- "Failed log progress while closing, ignoring and continuing shutdown. Message={}",
- e.getMessage());
- }
-
- // Notify and interrupt the waiting scheduler thread
- synchronized (this) {
- notifyAll();
- }
- // Interrupt the ShuffleScheduler thread only if the close is invoked by another thread.
- // If this is invoked on the same thread, then the shuffleRunner has already complete, and
- // there's
- // no point interrupting it.
- // The interrupt is needed to unblock any merges or waits which may be happening, so that
- // the thread can
- // exit.
- if (shuffleSchedulerThread != null
- && !Thread.currentThread().equals(shuffleSchedulerThread)) {
- shuffleSchedulerThread.interrupt();
- }
-
- // Interrupt the fetchers.
- for (CelebornTezShuffleDataFetcher fetcher : celebornRunningFetchers) {
- try {
- fetcher.shutDown();
- } catch (Exception e) {
- LOG.warn(
- "Error while shutting down fetcher. Ignoring and continuing shutdown. Message={}",
- e.getMessage());
- }
- }
-
- // Kill the Referee thread.
- try {
- referee.interrupt();
- referee.join();
- } catch (InterruptedException e) {
- LOG.warn("Interrupted while shutting down referee. Ignoring and continuing shutdown");
- Thread.currentThread().interrupt();
- } catch (Exception e) {
- LOG.warn(
- "Error while shutting down referee. Ignoring and continuing shutdown. Message={}",
- e.getMessage());
- }
- }
- } finally {
- long startTime = System.currentTimeMillis();
- if (!fetcherExecutor.isShutdown()) {
- // Ensure that fetchers respond to cancel request.
- fetcherExecutor.shutdownNow();
- }
- long endTime = System.currentTimeMillis();
- LOG.info(
- "Shutting down fetchers for input: {}, shutdown timetaken: {} ms, "
- + "hasFetcherExecutorStopped: {}",
- srcNameTrimmed,
- (endTime - startTime),
- hasFetcherExecutorStopped());
- }
- }
-
- @VisibleForTesting
- boolean hasFetcherExecutorStopped() {
- return fetcherExecutor.isShutdown();
- }
-
- @VisibleForTesting
- public boolean isShutdown() {
- return isShutdown.get();
- }
-
- protected synchronized void updateEventReceivedTime() {
- long relativeTime = System.currentTimeMillis() - startTime;
- if (firstEventReceived.getValue() == 0) {
- firstEventReceived.setValue(relativeTime);
- lastEventReceived.setValue(relativeTime);
- return;
- }
- lastEventReceived.setValue(relativeTime);
- }
-
- /**
- * Placeholder for tracking shuffle events in case we get multiple spills info for the same
- * attempt.
- */
- static class ShuffleEventInfo {
- BitSet eventsProcessed;
- int finalEventId = -1; // 0 indexed
- int attemptNum;
- String id;
- boolean scheduledForDownload; // whether chunks got scheduled for download (getMapHost)
-
- ShuffleEventInfo(InputAttemptIdentifier input) {
- this.id = input.getInputIdentifier() + "_" + input.getAttemptNumber();
- this.eventsProcessed = new BitSet();
- this.attemptNum = input.getAttemptNumber();
- }
-
- void spillProcessed(int spillId) {
- if (finalEventId != -1) {
- Preconditions.checkState(
- eventsProcessed.cardinality() <= (finalEventId + 1),
- "Wrong state. eventsProcessed cardinality="
- + eventsProcessed.cardinality()
- + " "
- + "finalEventId="
- + finalEventId
- + ", spillId="
- + spillId
- + ", "
- + toString());
- }
- eventsProcessed.set(spillId);
- }
-
- void setFinalEventId(int spillId) {
- finalEventId = spillId;
- }
-
- boolean isDone() {
- return ((finalEventId != -1) && (finalEventId + 1) == eventsProcessed.cardinality());
- }
-
- public String toString() {
- return "[eventsProcessed="
- + eventsProcessed
- + ", finalEventId="
- + finalEventId
- + ", id="
- + id
- + ", attemptNum="
- + attemptNum
- + ", scheduledForDownload="
- + scheduledForDownload
- + "]";
- }
- }
-
- public synchronized void copySucceeded(
- InputAttemptIdentifier srcAttemptIdentifier,
- MapHost host,
- long bytesCompressed,
- long bytesDecompressed,
- long millis,
- MapOutput output,
- boolean isLocalFetch)
- throws IOException {
-
- inputContext.notifyProgress();
- if (!isInputFinished(srcAttemptIdentifier.getInputIdentifier())) {
- if (!isLocalFetch) {
- /** Reset it only when it is a non-local-disk copy. */
- failedShufflesSinceLastCompletion = 0;
- }
- if (output != null) {
-
- failureCounts.remove(srcAttemptIdentifier);
- if (host != null) {
- hostFailures.remove(new HostPort(host.getHost(), host.getPort()));
- }
-
- output.commit();
- fetchStatsLogger.logIndividualFetchComplete(
- millis,
- bytesCompressed,
- bytesDecompressed,
- output.getType().toString(),
- srcAttemptIdentifier);
- if (output.getType() == Type.DISK) {
- bytesShuffledToDisk.increment(bytesCompressed);
- } else if (output.getType() == Type.DISK_DIRECT) {
- bytesShuffledToDiskDirect.increment(bytesCompressed);
- } else {
- bytesShuffledToMem.increment(bytesCompressed);
- }
- shuffledInputsCounter.increment(1);
- } else {
- // Output null implies that a physical input completion is being
- // registered without needing to fetch data
- skippedInputCounter.increment(1);
- }
-
- /**
- * In case of pipelined shuffle, it is quite possible that fetchers pulled the FINAL_UPDATE
- * spill in advance due to smaller output size. In such scenarios, we need to wait until we
- * retrieve all spill details to claim success.
- */
- if (!srcAttemptIdentifier.canRetrieveInputInChunks()) {
- remainingMaps.decrementAndGet();
- setInputFinished(srcAttemptIdentifier.getInputIdentifier());
- numFetchedSpills++;
- } else {
- int inputIdentifier = srcAttemptIdentifier.getInputIdentifier();
- // Allow only one task attempt to proceed.
- if (!validateInputAttemptForPipelinedShuffle(srcAttemptIdentifier)) {
- return;
- }
-
- ShuffleEventInfo eventInfo = pipelinedShuffleInfoEventsMap.get(inputIdentifier);
-
- // Possible that Shuffle event handler invoked this, due to empty partitions
- if (eventInfo == null && output == null) {
- eventInfo = new ShuffleEventInfo(srcAttemptIdentifier);
- pipelinedShuffleInfoEventsMap.put(inputIdentifier, eventInfo);
- }
-
- assert (eventInfo != null);
- eventInfo.spillProcessed(srcAttemptIdentifier.getSpillEventId());
- numFetchedSpills++;
-
- if (srcAttemptIdentifier.getFetchTypeInfo()
- == InputAttemptIdentifier.SPILL_INFO.FINAL_UPDATE) {
- eventInfo.setFinalEventId(srcAttemptIdentifier.getSpillEventId());
- }
-
- // check if we downloaded all spills pertaining to this InputAttemptIdentifier
- if (eventInfo.isDone()) {
- remainingMaps.decrementAndGet();
- setInputFinished(inputIdentifier);
- pipelinedShuffleInfoEventsMap.remove(inputIdentifier);
- if (LOG.isTraceEnabled()) {
- LOG.trace(
- "Removing : "
- + srcAttemptIdentifier
- + ", pending: "
- + pipelinedShuffleInfoEventsMap);
- }
- }
-
- if (LOG.isTraceEnabled()) {
- LOG.trace("eventInfo " + eventInfo.toString());
- }
- }
-
- if (remainingMaps.get() == 0) {
- notifyAll(); // Notify the getHost() method.
- LOG.info(
- "All inputs fetched for input vertex : " + inputContext.getInputOutputVertexNames());
- }
-
- // update the status
- lastProgressTime = System.currentTimeMillis();
- totalBytesShuffledTillNow += bytesCompressed;
- logProgress();
- reduceShuffleBytes.increment(bytesCompressed);
- reduceBytesDecompressed.increment(bytesDecompressed);
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "src task: "
- + TezRuntimeUtils.getTaskAttemptIdentifier(
- inputContext.getSourceVertexName(),
- srcAttemptIdentifier.getInputIdentifier(),
- srcAttemptIdentifier.getAttemptNumber())
- + " done");
- }
- } else {
- // input is already finished. duplicate fetch.
- LOG.warn("Duplicate fetch of input no longer needs to be fetched: " + srcAttemptIdentifier);
- // free the resource - specially memory
-
- // If the src does not generate data, output will be null.
- if (output != null) {
- output.abort();
- }
- }
- // TODO NEWTEZ Should this be releasing the output, if not committed ? Possible memory leak in
- // case of speculation.
- }
-
- private boolean validateInputAttemptForPipelinedShuffle(InputAttemptIdentifier input) {
- // For pipelined shuffle.
- // TODO: TEZ-2132 for error handling. As of now, fail fast if there is a different attempt
- if (input.canRetrieveInputInChunks()) {
- ShuffleEventInfo eventInfo = pipelinedShuffleInfoEventsMap.get(input.getInputIdentifier());
- if (eventInfo != null && input.getAttemptNumber() != eventInfo.attemptNum) {
- /*
- * Check if current attempt has been scheduled for download.
- * e.g currentAttemptNum=0, eventsProcessed={}, newAttemptNum=1
- * If nothing is scheduled in current attempt and no events are processed
- * (i.e copySucceeded), we can ignore current attempt and start processing the new
- * attempt (e.g LLAP).
- */
- if (eventInfo.scheduledForDownload || !eventInfo.eventsProcessed.isEmpty()) {
- IOException exception =
- new IOException(
- "Previous event already got scheduled for "
- + input
- + ". Previous attempt's data could have been already merged "
- + "to memory/disk outputs. Killing (self) this task early."
- + " currentAttemptNum="
- + eventInfo.attemptNum
- + ", eventsProcessed="
- + eventInfo.eventsProcessed
- + ", scheduledForDownload="
- + eventInfo.scheduledForDownload
- + ", newAttemptNum="
- + input.getAttemptNumber());
- String message = "Killing self as previous attempt data could have been consumed";
- killSelf(exception, message);
- return false;
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Ignoring current attempt="
- + eventInfo.attemptNum
- + " with eventInfo="
- + eventInfo.toString()
- + "and processing new attempt="
- + input.getAttemptNumber());
- }
- }
-
- if (eventInfo == null) {
- pipelinedShuffleInfoEventsMap.put(input.getInputIdentifier(), new ShuffleEventInfo(input));
- }
- }
- return true;
- }
-
- @VisibleForTesting
- void killSelf(Exception exception, String message) {
- LOG.error(message, exception);
- exceptionReporter.killSelf(exception, message);
- }
-
- private final AtomicInteger nextProgressLineEventCount = new AtomicInteger(0);
-
- private void logProgress() {
- int inputsDone = numInputs - remainingMaps.get();
- 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(
- InputAttemptFetchFailure fetchFailure,
- MapHost host,
- boolean readError,
- boolean connectError) {
- failedShuffleCounter.increment(1);
- inputContext.notifyProgress();
- int failures = incrementAndGetFailureAttempt(fetchFailure.getInputAttemptIdentifier());
-
- if (!fetchFailure.isLocalFetch()) {
- /**
- * Track the number of failures that has happened since last completion. This gets reset on a
- * successful copy.
- */
- failedShufflesSinceLastCompletion++;
- }
-
- /**
- * Inform AM: - In case of read/connect error - In case attempt failures exceed threshold of
- * maxFetchFailuresBeforeReporting (5) Bail-out if needed: - Check whether individual attempt
- * crossed failure threshold limits - Check overall shuffle health. Bail out if needed.*
- */
-
- // TEZ-2890
- boolean shouldInformAM =
- (reportReadErrorImmediately && (readError || connectError))
- || ((failures % maxFetchFailuresBeforeReporting) == 0);
-
- if (shouldInformAM) {
- // Inform AM. In case producer needs to be restarted, it is handled at AM.
- informAM(fetchFailure);
- }
-
- // Restart consumer in case shuffle is not healthy
- if (!isShuffleHealthy(fetchFailure)) {
- return;
- }
-
- penalizeHost(host, failures);
- }
-
- private boolean isAbortLimitExceeedFor(InputAttemptIdentifier srcAttempt) {
- int attemptFailures = getFailureCount(srcAttempt);
- if (attemptFailures >= abortFailureLimit) {
- // This task has seen too many fetch failures - report it as failed. The
- // AM may retry it if max failures has not been reached.
-
- // Between the task and the AM - someone needs to determine who is at
- // fault. If there's enough errors seen on the task, before the AM informs
- // it about source failure, the task considers itself to have failed and
- // allows the AM to re-schedule it.
- String errorMsg =
- "Failed "
- + attemptFailures
- + " times trying to "
- + "download from "
- + TezRuntimeUtils.getTaskAttemptIdentifier(
- inputContext.getSourceVertexName(),
- srcAttempt.getInputIdentifier(),
- srcAttempt.getAttemptNumber())
- + ". threshold="
- + abortFailureLimit;
- IOException ioe = new IOException(errorMsg);
- // Shuffle knows how to deal with failures post shutdown via the onFailure hook
- exceptionReporter.reportException(ioe);
- return true;
- }
- return false;
- }
-
- private void penalizeHost(MapHost host, int failures) {
- host.penalize();
-
- HostPort hostPort = new HostPort(host.getHost(), host.getPort());
- // TODO TEZ-922 hostFailures isn't really used for anything apart from
- // hasFailedAcrossNodes().Factor it into error
- // reporting / potential blacklisting of hosts.
- if (hostFailures.containsKey(hostPort)) {
- IntWritable x = hostFailures.get(hostPort);
- x.set(x.get() + 1);
- } else {
- hostFailures.put(hostPort, new IntWritable(1));
- }
-
- long delay = (long) (INITIAL_PENALTY * Math.pow(PENALTY_GROWTH_RATE, failures));
- long penaltyDelay = Math.min(delay, maxPenaltyTime);
- penalties.add(new Penalty(host, penaltyDelay));
- }
-
- private int getFailureCount(InputAttemptIdentifier srcAttempt) {
- IntWritable failureCount = failureCounts.get(srcAttempt);
- return (failureCount == null) ? 0 : failureCount.get();
- }
-
- private int incrementAndGetFailureAttempt(InputAttemptIdentifier srcAttempt) {
- int failures = 1;
- if (failureCounts.containsKey(srcAttempt)) {
- IntWritable x = failureCounts.get(srcAttempt);
- x.set(x.get() + 1);
- failures = x.get();
- } else {
- failureCounts.put(srcAttempt, new IntWritable(1));
- }
- return failures;
- }
-
- public void reportLocalError(IOException 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);
- }
-
- // Notify AM
- private void informAM(InputAttemptFetchFailure fetchFailure) {
- InputAttemptIdentifier srcAttempt = fetchFailure.getInputAttemptIdentifier();
- LOG.info(
- "{}: Reporting fetch failure for InputIdentifier: {} taskAttemptIdentifier: {}, "
- + "local fetch: {}, remote fetch failure reported as local failure: {}) to AM.",
- srcNameTrimmed,
- srcAttempt,
- TezRuntimeUtils.getTaskAttemptIdentifier(
- inputContext.getSourceVertexName(),
- srcAttempt.getInputIdentifier(),
- srcAttempt.getAttemptNumber()),
- fetchFailure.isLocalFetch(),
- fetchFailure.isDiskErrorAtSource());
- List<Event> failedEvents = Lists.newArrayListWithCapacity(1);
- failedEvents.add(
- InputReadErrorEvent.create(
- "Fetch failure for "
- + TezRuntimeUtils.getTaskAttemptIdentifier(
- inputContext.getSourceVertexName(),
- srcAttempt.getInputIdentifier(),
- srcAttempt.getAttemptNumber())
- + " to jobtracker.",
- srcAttempt.getInputIdentifier(),
- srcAttempt.getAttemptNumber(),
- fetchFailure.isLocalFetch(),
- fetchFailure.isDiskErrorAtSource(),
- localHostname));
-
- inputContext.sendEvents(failedEvents);
- }
-
- /**
- * To determine if failures happened across nodes or not. This will help in determining whether
- * this task needs to be restarted or source needs to be restarted.
- *
- * @param logContext context info for logging
- * @return boolean true indicates this task needs to be restarted
- */
- private boolean hasFailedAcrossNodes(String logContext) {
- int numUniqueHosts = uniqueHosts.size();
- Preconditions.checkArgument(numUniqueHosts > 0, "No values in unique hosts");
- int threshold = Math.max(3, (int) Math.ceil(numUniqueHosts * hostFailureFraction));
- int total = 0;
- boolean failedAcrossNodes = false;
- for (HostPort host : uniqueHosts) {
- IntWritable failures = hostFailures.get(host);
- if (failures != null && failures.get() > minFailurePerHost) {
- total++;
- failedAcrossNodes = (total > (threshold * minFailurePerHost));
- if (failedAcrossNodes) {
- break;
- }
- }
- }
-
- LOG.info(
- logContext
- + ", numUniqueHosts="
- + numUniqueHosts
- + ", hostFailureThreshold="
- + threshold
- + ", hostFailuresCount="
- + hostFailures.size()
- + ", hosts crossing threshold="
- + total
- + ", reducerFetchIssues="
- + failedAcrossNodes);
-
- return failedAcrossNodes;
- }
-
- private boolean allEventsReceived() {
- if (!pipelinedShuffleInfoEventsMap.isEmpty()) {
- return (pipelinedShuffleInfoEventsMap.size() == numInputs);
- } else {
- // no pipelining
- return ((pathToIdentifierMap.size() + skippedInputCounter.getValue()) == numInputs);
- }
+ private boolean allInputTaskAttemptDone() {
+ return (this.partitionIdToSuccessTezTasks.values().stream().mapToInt(s -> s.size()).sum()
+ + skippedInputCounter.getValue())
+ == numInputs;
}
private boolean isAllInputFetched() {
- return allEventsReceived() && (successRssPartitionSet.size() >= allRssPartition.size());
- }
-
- /**
- * Check if consumer needs to be restarted based on total failures w.r.t completed outputs and
- * based on number of errors that have happened since last successful completion. Consider into
- * account whether failures have been seen across different nodes.
- *
- * @return true to indicate fetchers are healthy
- */
- private boolean isFetcherHealthy(String logContext) {
-
- long totalFailures = failedShuffleCounter.getValue();
- int doneMaps = numInputs - remainingMaps.get();
-
- boolean fetcherHealthy = true;
- if (doneMaps > 0) {
- fetcherHealthy =
- (((float) totalFailures / (totalFailures + doneMaps)) < maxAllowedFailedFetchFraction);
- }
-
- if (fetcherHealthy) {
- // Compute this logic only when all events are received
- if (allEventsReceived()) {
- if (hostFailureFraction > 0) {
- boolean failedAcrossNodes = hasFailedAcrossNodes(logContext);
- if (failedAcrossNodes) {
- return false; // not healthy
- }
- }
-
- if (checkFailedFetchSinceLastCompletion) {
- /**
- * remainingMaps works better instead of pendingHosts in the following condition because
- * of the way the fetcher reports failures
- */
- if (failedShufflesSinceLastCompletion >= remainingMaps.get() * minFailurePerHost) {
- /**
- * Check if lots of errors are seen after last progress time.
- *
- * <p>E.g totalFailures = 20. doneMaps = 320 - 300; fetcherHealthy = (20/(20+300)) <
- * 0.5. So reducer would be marked as healthy. Assume 20 errors happen when downloading
- * the last 20 attempts. Host failure & individual attempt failures would keep
- * increasing; but at very slow rate 15 * 180 seconds per attempt to find out the issue.
- *
- * <p>Instead consider the new errors with the pending items to be fetched. Assume 21
- * new errors happened after last progress; remainingMaps = (320-300) = 20; (21 / (21 +
- * 20)) > 0.5 So we reset the reducer to unhealthy here (special case)
- *
- * <p>In normal conditions (i.e happy path), this wouldn't even cause any issue as
- * failedShufflesSinceLastCompletion is reset as soon as we see successful download.
- */
- fetcherHealthy =
- (((float) failedShufflesSinceLastCompletion
- / (failedShufflesSinceLastCompletion + remainingMaps.get()))
- < maxAllowedFailedFetchFraction);
-
- LOG.info(
- logContext
- + ", fetcherHealthy="
- + fetcherHealthy
- + ", failedShufflesSinceLastCompletion="
- + failedShufflesSinceLastCompletion
- + ", remainingMaps="
- + remainingMaps.get());
- }
- }
- }
- }
- return fetcherHealthy;
- }
-
- boolean isShuffleHealthy(InputAttemptFetchFailure fetchFailure) {
- InputAttemptIdentifier srcAttempt = fetchFailure.getInputAttemptIdentifier();
- if (isAbortLimitExceeedFor(srcAttempt)) {
- return false;
- }
-
- final float MIN_REQUIRED_PROGRESS_PERCENT = minReqProgressFraction;
- final float MAX_ALLOWED_STALL_TIME_PERCENT = maxStallTimeFraction;
-
- int doneMaps = numInputs - remainingMaps.get();
-
- String logContext = "srcAttempt=" + srcAttempt.toString();
- boolean fetcherHealthy = isFetcherHealthy(logContext);
-
- // check if the reducer has progressed enough
- boolean reducerProgressedEnough =
- (((float) doneMaps / numInputs) >= MIN_REQUIRED_PROGRESS_PERCENT);
-
- // check if the reducer is stalled for a long time
- // duration for which the reducer is stalled
- int stallDuration = (int) (System.currentTimeMillis() - lastProgressTime);
-
- // duration for which the reducer ran with progress
- int shuffleProgressDuration = (int) (lastProgressTime - startTime);
-
- boolean reducerStalled =
- (shuffleProgressDuration > 0)
- && (((float) stallDuration / shuffleProgressDuration)
- >= MAX_ALLOWED_STALL_TIME_PERCENT);
-
- // kill if not healthy and has insufficient progress
- if ((failureCounts.size() >= maxFailedUniqueFetches
- || failureCounts.size() == (numInputs - doneMaps))
- && !fetcherHealthy
- && (!reducerProgressedEnough || reducerStalled)) {
- String errorMsg =
- (srcNameTrimmed
- + ": "
- + "Shuffle failed with too many fetch failures and insufficient progress: "
- + "[failureCounts="
- + failureCounts.size()
- + ", pendingInputs="
- + (numInputs - doneMaps)
- + ", fetcherHealthy="
- + fetcherHealthy
- + ", reducerProgressedEnough="
- + reducerProgressedEnough
- + ", reducerStalled="
- + reducerStalled
- + ", hostFailures="
- + hostFailures)
- + "]";
- LOG.error(errorMsg);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Host failures=" + hostFailures.keySet());
- }
- // Shuffle knows how to deal with failures post shutdown via the onFailure hook
- exceptionReporter.reportException(new IOException(errorMsg, fetchFailure.getCause()));
- return false;
- }
- return true;
+ return allInputTaskAttemptDone() && (successRssPartitionSet.size() >= allRssPartition.size());
}
public synchronized void addKnownMapOutput(
String inputHostName, int port, int partitionId, CompositeInputAttemptIdentifier srcAttempt) {
allRssPartition.add(partitionId);
- if (!partitionIdToSuccessMapTaskAttempts.containsKey(partitionId)) {
- partitionIdToSuccessMapTaskAttempts.put(partitionId, new HashSet<>());
- }
- partitionIdToSuccessMapTaskAttempts.get(partitionId).add(srcAttempt);
-
- uniqueHosts.add(new HostPort(inputHostName, port));
- HostPortPartition identifier = new HostPortPartition(inputHostName, port, partitionId);
-
- MapHost host = mapLocations.get(identifier);
- if (host == null) {
- host = new MapHost(inputHostName, port, partitionId, srcAttempt.getInputIdentifierCount());
- mapLocations.put(identifier, host);
- }
-
- // Allow only one task attempt to proceed.
- if (!validateInputAttemptForPipelinedShuffle(srcAttempt)) {
- return;
- }
-
- host.addKnownMap(srcAttempt);
- for (int i = 0; i < srcAttempt.getInputIdentifierCount(); i++) {
- PathPartition pathPartition =
- new PathPartition(srcAttempt.getPathComponent(), partitionId + i);
- pathToIdentifierMap.put(pathPartition, srcAttempt.expand(i));
- }
-
- // Mark the host as pending
- if (host.getState() == MapHost.State.PENDING) {
- pendingHosts.add(host);
- notifyAll();
- }
- }
-
- public void obsoleteInput(InputAttemptIdentifier srcAttempt) {
- // The incoming srcAttempt does not contain a path component.
- LOG.info(srcNameTrimmed + ": " + "Adding obsolete input: " + srcAttempt);
- ShuffleEventInfo eventInfo = pipelinedShuffleInfoEventsMap.get(srcAttempt.getInputIdentifier());
+ Set<InputAttemptIdentifier> inputAttemptIdentifiers =
+ partitionIdToSuccessMapTaskAttempts.computeIfAbsent(partitionId, id -> new HashSet<>());
+ String pathComponent = srcAttempt.getPathComponent();
+ TezTaskAttemptID tezTaskAttemptId =
+ TezTaskAttemptID.fromString(pathComponent.substring(0, pathComponent.length() - 6));
+ partitionIdToSuccessTezTasks.putIfAbsent(partitionId, new HashSet<>());
+ partitionIdToSuccessTezTasks.get(partitionId).add(tezTaskAttemptId.getTaskID());
- // Pipelined shuffle case (where pipelinedShuffleInfoEventsMap gets populated).
- // Fail fast here.
- if (eventInfo != null) {
- // In case this we haven't started downloading it, get rid of it.
- if (eventInfo.eventsProcessed.isEmpty() && !eventInfo.scheduledForDownload) {
- // obsoleted anyways; no point tracking if nothing is started
- pipelinedShuffleInfoEventsMap.remove(srcAttempt.getInputIdentifier());
- LOG.debug("Removing {} from tracking", eventInfo);
- return;
- }
- IOException exception =
- new IOException(
- srcAttempt
- + " is marked as obsoleteInput, but it "
- + "exists in shuffleInfoEventMap. Some data could have been already merged "
- + "to memory/disk outputs. Failing the fetch early. eventInfo: "
- + eventInfo);
- String message =
- "Got obsolete event. Killing self as attempt's data could have been consumed";
- killSelf(exception, message);
- return;
- }
- synchronized (this) {
- obsoleteInputs.add(srcAttempt);
- }
- }
-
- public synchronized void putBackKnownMapOutput(MapHost host, InputAttemptIdentifier srcAttempt) {
- host.addKnownMap(srcAttempt);
+ inputAttemptIdentifiers.add(srcAttempt);
+ super.addKnownMapOutput(inputHostName, port, partitionId, srcAttempt);
}
public synchronized MapHost getHost() throws InterruptedException {
@@ -1273,252 +167,19 @@ class CelebornScheduler extends ShuffleScheduler {
pendingHosts.remove(host);
host.markBusy();
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- srcNameTrimmed
- + ": "
- + "Assigning "
- + host
- + " with "
- + host.getNumKnownMapOutputs()
- + " to "
- + Thread.currentThread().getName());
- }
- shuffleStart.set(System.currentTimeMillis());
return host;
} else {
return null;
}
}
- public InputAttemptIdentifier getIdentifierForFetchedOutput(String path, int reduceId) {
- return pathToIdentifierMap.get(new PathPartition(path, reduceId));
- }
-
- private synchronized boolean inputShouldBeConsumed(InputAttemptIdentifier id) {
- boolean isInputFinished = false;
- if (id instanceof CompositeInputAttemptIdentifier) {
- CompositeInputAttemptIdentifier cid = (CompositeInputAttemptIdentifier) id;
- isInputFinished =
- isInputFinished(
- cid.getInputIdentifier(), cid.getInputIdentifier() + cid.getInputIdentifierCount());
- } else {
- isInputFinished = isInputFinished(id.getInputIdentifier());
- }
- return !obsoleteInputs.contains(id) && !isInputFinished;
- }
-
- public synchronized List<InputAttemptIdentifier> getMapsForHost(MapHost host) {
- List<InputAttemptIdentifier> origList = host.getAndClearKnownMaps();
-
- ListMultimap<Integer, InputAttemptIdentifier> dedupedList = LinkedListMultimap.create();
-
- Iterator<InputAttemptIdentifier> listItr = origList.iterator();
- while (listItr.hasNext()) {
- // we may want to try all versions of the input but with current retry
- // behavior older ones are likely to be lost and should be ignored.
- // This may be removed after TEZ-914
- InputAttemptIdentifier id = listItr.next();
- if (inputShouldBeConsumed(id)) {
- Integer inputNumber = Integer.valueOf(id.getInputIdentifier());
- List<InputAttemptIdentifier> oldIdList = dedupedList.get(inputNumber);
-
- if (oldIdList == null || oldIdList.isEmpty()) {
- dedupedList.put(inputNumber, id);
- continue;
- }
-
- // In case of pipelined shuffle, we can have multiple spills. In such cases, we can have
- // more than one item in the oldIdList.
- boolean addIdentifierToList = false;
- Iterator<InputAttemptIdentifier> oldIdIterator = oldIdList.iterator();
- while (oldIdIterator.hasNext()) {
- InputAttemptIdentifier oldId = oldIdIterator.next();
-
- // no need to add if spill ids are same
- if (id.canRetrieveInputInChunks()) {
- if (oldId.getSpillEventId() == id.getSpillEventId()) {
- // TODO: need to handle deterministic spills later.
- addIdentifierToList = false;
- continue;
- } else if (oldId.getAttemptNumber() == id.getAttemptNumber()) {
- // but with different spill id.
- addIdentifierToList = true;
- break;
- }
- }
-
- // if its from different attempt, take the latest attempt
- if (oldId.getAttemptNumber() < id.getAttemptNumber()) {
- // remove existing identifier
- oldIdIterator.remove();
- LOG.warn(
- "Old Src for InputIndex: "
- + inputNumber
- + " with attemptNumber: "
- + oldId.getAttemptNumber()
- + " was not determined to be invalid. Ignoring it for now in favour of "
- + id.getAttemptNumber());
- addIdentifierToList = true;
- break;
- }
- }
- if (addIdentifierToList) {
- dedupedList.put(inputNumber, id);
- }
- } else {
- LOG.info("Ignoring finished or obsolete source: " + id);
- }
- }
-
- // Compute the final list, limited by NUM_FETCHERS_AT_ONCE
- List<InputAttemptIdentifier> result = new ArrayList<InputAttemptIdentifier>();
- int includedMaps = 0;
- int totalSize = dedupedList.size();
-
- for (Integer inputIndex : dedupedList.keySet()) {
- List<InputAttemptIdentifier> attemptIdentifiers = dedupedList.get(inputIndex);
- for (InputAttemptIdentifier inputAttemptIdentifier : attemptIdentifiers) {
- if (includedMaps++ >= maxTaskOutputAtOnce) {
- host.addKnownMap(inputAttemptIdentifier);
- } else {
- if (inputAttemptIdentifier.canRetrieveInputInChunks()) {
- ShuffleEventInfo shuffleEventInfo =
- pipelinedShuffleInfoEventsMap.get(inputAttemptIdentifier.getInputIdentifier());
- if (shuffleEventInfo != null) {
- shuffleEventInfo.scheduledForDownload = true;
- }
- }
- result.add(inputAttemptIdentifier);
- }
- }
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "assigned "
- + includedMaps
- + " of "
- + totalSize
- + " to "
- + host
- + " to "
- + Thread.currentThread().getName());
- }
- return result;
- }
-
- public synchronized void freeHost(MapHost host) {
- if (host.getState() != MapHost.State.PENALIZED) {
- if (host.markAvailable() == MapHost.State.PENDING) {
- pendingHosts.add(host);
- notifyAll();
- }
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- host
- + " freed by "
- + Thread.currentThread().getName()
- + " in "
- + (System.currentTimeMillis() - shuffleStart.get())
- + "ms");
- }
- }
-
- public synchronized void resetKnownMaps() {
- mapLocations.clear();
- obsoleteInputs.clear();
- pendingHosts.clear();
- pathToIdentifierMap.clear();
- }
-
- /**
- * Utility method to check if the Shuffle data fetch is complete.
- *
- * @return true if complete
- */
- public synchronized boolean isDone() {
- return remainingMaps.get() == 0;
- }
-
- /** A structure that records the penalty for a host. */
- static class Penalty implements Delayed {
- MapHost host;
- private long endTime;
-
- Penalty(MapHost host, long delay) {
- this.host = host;
- this.endTime = System.currentTimeMillis() + delay;
- }
-
- public long getDelay(TimeUnit unit) {
- long remainingTime = endTime - System.currentTimeMillis();
- return unit.convert(remainingTime, TimeUnit.MILLISECONDS);
- }
-
- public int compareTo(Delayed o) {
- long other = ((Penalty) o).endTime;
- return endTime == other ? 0 : (endTime < other ? -1 : 1);
- }
- }
-
- /** A thread that takes hosts off of the penalty list when the timer expires. */
- private class Referee extends Thread {
- public Referee() {
- setName(
- "ShufflePenaltyReferee {"
- + TezUtilsInternal.cleanVertexName(inputContext.getSourceVertexName())
- + "}");
- setDaemon(true);
- }
-
- public void run() {
- try {
- while (!isShutdown.get()) {
- // take the first host that has an expired penalty
- MapHost host = penalties.take().host;
- synchronized (CelebornScheduler.this) {
- if (host.markAvailable() == MapHost.State.PENDING) {
- pendingHosts.add(host);
- CelebornScheduler.this.notifyAll();
- }
- }
- }
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- // This handles shutdown of the entire fetch / merge process.
- } catch (Throwable t) {
- // Shuffle knows how to deal with failures post shutdown via the onFailure hook
- exceptionReporter.reportException(t);
- }
- }
- }
-
- void setInputFinished(int inputIndex) {
- synchronized (finishedMaps) {
- finishedMaps.set(inputIndex, true);
- }
- }
-
- boolean isInputFinished(int inputIndex) {
- synchronized (finishedMaps) {
- return finishedMaps.get(inputIndex);
- }
- }
-
- boolean isInputFinished(int inputIndex, int inputEnd) {
- synchronized (finishedMaps) {
- return finishedMaps.nextClearBit(inputIndex) > inputEnd;
- }
- }
-
private class CelebornShuffleSchedulerCallable extends CallableWithNdc<Void> {
@Override
protected Void callInternal() throws InterruptedException {
while (!isShutdown.get() && !isAllInputFetched()) {
synchronized (CelebornScheduler.this) {
- while (!allEventsReceived()
+ while (!allInputTaskAttemptDone()
|| ((celebornRunningFetchers.size() >= numFetchers || pendingHosts.isEmpty())
&& !isAllInputFetched())) {
try {
@@ -1539,11 +200,6 @@ class CelebornScheduler extends ShuffleScheduler {
}
}
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- srcNameTrimmed + ": " + "NumCompletedInputs: {}" + (numInputs - remainingMaps.get()));
- }
-
// Ensure there's memory available before scheduling the next Fetcher.
try {
// If merge is on, block
@@ -1631,7 +287,8 @@ class CelebornScheduler extends ShuffleScheduler {
private synchronized boolean isFirstRssPartitionFetch(MapHost mapHost) {
Integer partitionId = mapHost.getPartitionId();
- LOG.info("Check isFirstRssPartitionFetch, mapHost:{},partitionId:{}", mapHost, partitionId);
+ LOG.info(
+ "Check isFirstCelebornPartitionFetch, mapHost:{},partitionId:{}", mapHost, partitionId);
if (runningRssPartitionMap.containsKey(partitionId)
|| successRssPartitionSet.contains(partitionId)) {
@@ -1646,15 +303,13 @@ class CelebornScheduler extends ShuffleScheduler {
CelebornTezReader reader =
new CelebornTezReader(
shuffleClient, shuffleId, partitionId, applicationAttemptId.getAttemptId());
- CelebornTezShuffleDataFetcher fetcher =
- new CelebornTezShuffleDataFetcher(
- partitionIdToSuccessMapTaskAttempts.get(mapHost.getPartitionId()).iterator().next(),
- mapHost.getPartitionId(),
- mergeManager,
- inputContext.getCounters(),
- reader,
- exceptionReporter);
- return fetcher;
+ return new CelebornTezShuffleDataFetcher(
+ partitionIdToSuccessMapTaskAttempts.get(mapHost.getPartitionId()).iterator().next(),
+ mapHost.getPartitionId(),
+ mergeManager,
+ inputContext.getCounters(),
+ reader,
+ exceptionReporter);
}
}
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornShuffle.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornShuffle.java
index 9b34211d7..a1a15bf0f 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornShuffle.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornShuffle.java
@@ -20,88 +20,19 @@ package org.apache.tez.runtime.library.common.shuffle.orderedgrouped;
import static org.apache.celeborn.tez.plugin.util.CelebornTezUtils.*;
import java.io.IOException;
-import java.lang.reflect.UndeclaredThrowableException;
-import java.util.List;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.ListeningExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.tez.common.CallableWithNdc;
-import org.apache.tez.common.GuavaShim;
-import org.apache.tez.common.Preconditions;
-import org.apache.tez.common.TezRuntimeFrameworkConfigs;
-import org.apache.tez.common.TezUtilsInternal;
-import org.apache.tez.common.counters.TaskCounter;
-import org.apache.tez.common.counters.TezCounter;
-import org.apache.tez.dag.api.TezException;
-import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.InputContext;
-import org.apache.tez.runtime.api.TaskFailureType;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.apache.tez.runtime.library.common.TezRuntimeUtils;
-import org.apache.tez.runtime.library.common.combine.Combiner;
import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
-import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
-import org.apache.tez.runtime.library.exceptions.InputAlreadyClosedException;
-import org.apache.tez.runtime.library.utils.CodecUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.celeborn.client.ShuffleClient;
import org.apache.celeborn.common.CelebornConf;
import org.apache.celeborn.common.identity.UserIdentifier;
import org.apache.celeborn.tez.plugin.util.CelebornTezUtils;
-public class CelebornShuffle implements ExceptionReporter {
- private static final Logger LOG = LoggerFactory.getLogger(Shuffle.class);
-
- private final Configuration conf;
- private final InputContext inputContext;
-
- private final ShuffleInputEventHandlerOrderedGrouped eventHandler;
-
- final CelebornScheduler scheduler;
-
- @VisibleForTesting final MergeManager merger;
-
- private final CompressionCodec codec;
- private final boolean ifileReadAhead;
- private final int ifileReadAheadLength;
-
- private AtomicReference<Throwable> throwable = new AtomicReference<Throwable>();
- private String throwingThreadName = null;
-
- private final RunShuffleCallable runShuffleCallable;
- private volatile ListenableFuture<TezRawKeyValueIterator> runShuffleFuture;
- private final ListeningExecutorService executor;
-
- private final String sourceDestNameTrimmed;
-
- private AtomicBoolean isShutDown = new AtomicBoolean(false);
- private AtomicBoolean fetchersClosed = new AtomicBoolean(false);
- private AtomicBoolean schedulerClosed = new AtomicBoolean(false);
- private AtomicBoolean mergerClosed = new AtomicBoolean(false);
-
- private final TezCounter mergePhaseTime;
- private final TezCounter shufflePhaseTime;
-
- private final long startTime;
- private Configuration remoteConf;
-
+public class CelebornShuffle extends Shuffle {
public CelebornShuffle(
InputContext inputContext,
Configuration conf,
@@ -109,73 +40,32 @@ public class CelebornShuffle implements ExceptionReporter {
long initialMemoryAvailable,
ApplicationAttemptId applicationAttemptId)
throws IOException {
- this.inputContext = inputContext;
- this.conf = conf;
-
- this.sourceDestNameTrimmed =
- TezUtilsInternal.cleanVertexName(inputContext.getSourceVertexName())
- + " -> "
- + TezUtilsInternal.cleanVertexName(inputContext.getTaskVertexName());
-
- this.codec = CodecUtils.getCodec(conf);
-
- this.ifileReadAhead =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT);
- if (this.ifileReadAhead) {
- this.ifileReadAheadLength =
- conf.getInt(
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
- TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT);
- } else {
- this.ifileReadAheadLength = 0;
- }
-
- Combiner combiner = TezRuntimeUtils.instantiateCombiner(conf, inputContext);
-
- FileSystem localFS = FileSystem.getLocal(this.conf);
- LocalDirAllocator localDirAllocator =
- new LocalDirAllocator(TezRuntimeFrameworkConfigs.LOCAL_DIRS);
-
- // TODO TEZ Get rid of Map / Reduce references.
- TezCounter spilledRecordsCounter =
- inputContext.getCounters().findCounter(TaskCounter.SPILLED_RECORDS);
- TezCounter reduceCombineInputCounter =
- inputContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
- TezCounter mergedMapOutputsCounter =
- inputContext.getCounters().findCounter(TaskCounter.MERGED_MAP_OUTPUTS);
-
- startTime = System.currentTimeMillis();
- merger =
- new MergeManager(
- this.conf,
- localFS,
- localDirAllocator,
- inputContext,
- combiner,
- spilledRecordsCounter,
- reduceCombineInputCounter,
- mergedMapOutputsCounter,
- this,
- initialMemoryAvailable,
- codec,
- ifileReadAhead,
- ifileReadAheadLength);
+ super(inputContext, conf, numInputs, initialMemoryAvailable);
String host = conf.get(TEZ_CELEBORN_LM_HOST);
int port = conf.getInt(TEZ_CELEBORN_LM_PORT, -1);
int shuffleId = conf.getInt(TEZ_SHUFFLE_ID, -1);
String appId = conf.get(TEZ_CELEBORN_APPLICATION_ID);
- String user = conf.get(TEZ_CELEBORN_USER);
CelebornConf celebornConf = CelebornTezUtils.fromTezConfiguration(conf);
ShuffleClient shuffleClient =
- ShuffleClient.get(appId, host, port, celebornConf, UserIdentifier.apply(user), null);
-
- scheduler =
+ ShuffleClient.get(
+ appId,
+ host,
+ port,
+ celebornConf,
+ new UserIdentifier(
+ celebornConf.quotaUserSpecificTenant(), celebornConf.quotaUserSpecificUserName()),
+ null);
+
+ long startTime = (long) getParentPrivateField(this, "startTime");
+ CompressionCodec codec = (CompressionCodec) getParentPrivateField(this, "codec");
+ boolean ifileReadAhead = (boolean) getParentPrivateField(this, "ifileReadAhead");
+ int ifileReadAheadLength = (int) getParentPrivateField(this, "ifileReadAheadLength");
+ String sourceDestNameTrimmed = (String) getParentPrivateField(this, "sourceDestNameTrimmed");
+ CelebornScheduler scheduler =
new CelebornScheduler(
- this.inputContext,
- this.conf,
+ inputContext,
+ conf,
numInputs,
this,
shuffleClient,
@@ -188,247 +78,10 @@ public class CelebornShuffle implements ExceptionReporter {
sourceDestNameTrimmed,
shuffleId,
applicationAttemptId);
-
- this.mergePhaseTime = inputContext.getCounters().findCounter(TaskCounter.MERGE_PHASE_TIME);
- this.shufflePhaseTime = inputContext.getCounters().findCounter(TaskCounter.SHUFFLE_PHASE_TIME);
-
- eventHandler =
+ ShuffleInputEventHandlerOrderedGrouped eventHandler =
new ShuffleInputEventHandlerOrderedGrouped(
inputContext, scheduler, ShuffleUtils.isTezShuffleHandler(conf));
-
- ExecutorService rawExecutor =
- Executors.newFixedThreadPool(
- 1,
- new ThreadFactoryBuilder()
- .setDaemon(true)
- .setNameFormat("ShuffleAndMergeRunner {" + sourceDestNameTrimmed + "}")
- .build());
-
- executor = MoreExecutors.listeningDecorator(rawExecutor);
- runShuffleCallable = new RunShuffleCallable();
- }
-
- public void run() throws IOException {
- merger.configureAndStart();
- runShuffleFuture = executor.submit(runShuffleCallable);
- Futures.addCallback(
- runShuffleFuture, new ShuffleRunnerFutureCallback(), GuavaShim.directExecutor());
- executor.shutdown();
- }
-
- public void handleEvents(List<Event> events) throws IOException {
- if (!isShutDown.get()) {
- eventHandler.handleEvents(events);
- } else {
- LOG.info(
- sourceDestNameTrimmed
- + ": "
- + "Ignoring events since already shutdown. EventCount: "
- + events.size());
- }
- }
-
- public boolean isInputReady() throws IOException, InterruptedException, TezException {
- if (isShutDown.get()) {
- throw new InputAlreadyClosedException();
- }
- if (throwable.get() != null) {
- handleThrowable(throwable.get());
- }
- if (runShuffleFuture == null) {
- return false;
- }
- // Don't need to check merge status, since runShuffleFuture will only
- // complete once merge is complete.
- return runShuffleFuture.isDone();
- }
-
- public void shutdown() {
- if (!isShutDown.getAndSet(true)) {
- // Interrupt so that the scheduler / merger sees this interrupt.
- LOG.info("Shutting down Shuffle for source: " + sourceDestNameTrimmed);
- runShuffleFuture.cancel(true);
- cleanupIgnoreErrors();
- }
- }
-
- @Override
- public synchronized void reportException(Throwable t) {
- // RunShuffleCallable onFailure deals with ignoring errors on shutdown.
- if (throwable.get() == null) {
- LOG.info(
- sourceDestNameTrimmed
- + ": "
- + "Setting throwable in reportException with message ["
- + t.getMessage()
- + "] from thread ["
- + Thread.currentThread().getName());
- throwable.set(t);
- throwingThreadName = Thread.currentThread().getName();
- // Notify the scheduler so that the reporting thread finds the
- // exception immediately.
- cleanupShuffleSchedulerIgnoreErrors();
- }
- }
-
- private void cleanupShuffleSchedulerIgnoreErrors() {
- try {
- cleanupShuffleScheduler();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- LOG.info(
- sourceDestNameTrimmed
- + ": "
- + "Interrupted while attempting to close the scheduler during cleanup. Ignoring");
- }
- }
-
- private void cleanupShuffleScheduler() throws InterruptedException {
- if (!schedulerClosed.getAndSet(true)) {
- scheduler.close();
- }
- }
-
- @Override
- public void killSelf(Exception e, String s) {}
-
- public TezRawKeyValueIterator waitForInput()
- throws IOException, InterruptedException, TezException {
- Preconditions.checkState(runShuffleFuture != null, "waitForInput can only be called after run");
- TezRawKeyValueIterator kvIter = null;
- try {
- kvIter = runShuffleFuture.get();
- } catch (ExecutionException e) {
- Throwable cause = e.getCause();
- // Processor interrupted while waiting for errors, will see an InterruptedException.
- handleThrowable(cause);
- }
- if (isShutDown.get()) {
- throw new InputAlreadyClosedException();
- }
- if (throwable.get() != null) {
- handleThrowable(throwable.get());
- }
- return kvIter;
- }
-
- private void handleThrowable(Throwable t) throws IOException, InterruptedException {
- if (t instanceof IOException) {
- throw (IOException) t;
- } else if (t instanceof InterruptedException) {
- throw (InterruptedException) t;
- } else {
- throw new UndeclaredThrowableException(t);
- }
- }
-
- private class RunShuffleCallable extends CallableWithNdc<TezRawKeyValueIterator> {
- @Override
- protected TezRawKeyValueIterator callInternal() throws IOException, InterruptedException {
- if (!isShutDown.get()) {
- try {
- scheduler.start();
- } catch (Throwable e) {
- throw new Shuffle.ShuffleError("Error during shuffle", e);
- } finally {
- }
- }
- // The ShuffleScheduler may have exited cleanly as a result of a shutdown invocation
- // triggered by a previously reportedException. Check before proceeding further.s
- synchronized (CelebornShuffle.this) {
- if (throwable.get() != null) {
- throw new Shuffle.ShuffleError(
- "error in shuffle in " + throwingThreadName, throwable.get());
- }
- }
-
- // Finish the on-going merges...
- TezRawKeyValueIterator kvIter = null;
- inputContext.notifyProgress();
- try {
- kvIter = merger.close(true);
- } catch (Throwable e) {
- // Set the throwable so that future.get() sees the reported errror.
- throwable.set(e);
- throw new Shuffle.ShuffleError("Error while doing final merge ", e);
- }
- mergePhaseTime.setValue(System.currentTimeMillis() - startTime);
-
- inputContext.notifyProgress();
- // Sanity check
- synchronized (CelebornShuffle.this) {
- if (throwable.get() != null) {
- throw new Shuffle.ShuffleError(
- "error in shuffle in " + throwingThreadName, throwable.get());
- }
- }
-
- inputContext.inputIsReady();
- LOG.info("merge complete for input vertex : " + sourceDestNameTrimmed);
- return kvIter;
- }
- }
-
- private class ShuffleRunnerFutureCallback implements FutureCallback<TezRawKeyValueIterator> {
- @Override
- public void onSuccess(TezRawKeyValueIterator result) {
- LOG.info(sourceDestNameTrimmed + ": " + "Shuffle Runner thread complete");
- }
-
- @Override
- public void onFailure(Throwable t) {
- if (isShutDown.get()) {
- LOG.info(sourceDestNameTrimmed + ": " + "Already shutdown. Ignoring error");
- } else {
- LOG.error(sourceDestNameTrimmed + ": " + "ShuffleRunner failed with error", t);
- // In case of an abort / Interrupt - the runtime makes sure that this is ignored.
- inputContext.reportFailure(TaskFailureType.NON_FATAL, t, "Shuffle Runner Failed");
- cleanupIgnoreErrors();
- }
- }
- }
-
- private void cleanupIgnoreErrors() {
- try {
- if (eventHandler != null) {
- eventHandler.logProgress(true);
- }
- try {
- } catch (Exception e) {
- LOG.warn(
- "Error cleaning up shuffle scheduler. Ignoring and continuing with shutdown. Message={}",
- e.getMessage());
- }
- cleanupMerger(true);
- } catch (Throwable t) {
- LOG.info(sourceDestNameTrimmed + ": " + "Error in cleaning up.., ", t);
- }
- }
-
- private void cleanupMerger(boolean ignoreErrors) throws Throwable {
- if (!mergerClosed.getAndSet(true)) {
- try {
- merger.close(false);
- } catch (InterruptedException e) {
- if (ignoreErrors) {
- // Reset the status
- Thread.currentThread().interrupt();
- LOG.info(
- sourceDestNameTrimmed
- + ": "
- + "Interrupted while attempting to close the merger during cleanup. Ignoring");
- } else {
- throw e;
- }
- } catch (Throwable e) {
- if (ignoreErrors) {
- LOG.info(
- sourceDestNameTrimmed + ": " + "Exception while trying to shutdown merger, Ignoring",
- e);
- } else {
- throw e;
- }
- }
- }
+ setParentPrivateField(this, "scheduler", scheduler);
+ setParentPrivateField(this, "eventHandler", eventHandler);
}
}
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornTezShuffleDataFetcher.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornTezShuffleDataFetcher.java
index 97907d613..107915238 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornTezShuffleDataFetcher.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/CelebornTezShuffleDataFetcher.java
@@ -56,7 +56,6 @@ public class CelebornTezShuffleDataFetcher extends CallableWithNdc<Void> {
private long serializeTime = 0;
private long waitTime = 0;
private long copyTime = 0; // the sum of readTime + decompressTime + serializeTime + waitTime
- private long unCompressionLength = 0;
private final InputAttemptIdentifier inputAttemptIdentifier;
private static int uniqueMapId = 0;
@@ -86,7 +85,7 @@ public class CelebornTezShuffleDataFetcher extends CallableWithNdc<Void> {
this.celebornTezReader = celebornTezReader;
LOG.info(
- "RssTezShuffleDataFetcher, partitionId:{}, inputAttemptIdentifier:{}.",
+ "CelebornTezShuffleDataFetcher, partitionId:{}, inputAttemptIdentifier:{}.",
this.partitionId,
this.inputAttemptIdentifier);
}
@@ -167,9 +166,7 @@ public class CelebornTezShuffleDataFetcher extends CallableWithNdc<Void> {
+ readTime
+ " ms to fetch and "
+ decompressTime
- + " ms to decompress with unCompressionLength["
- + unCompressionLength
- + "] and "
+ + " ms to decompress "
+ serializeTime
+ " ms to serialize and "
+ waitTime
@@ -184,10 +181,6 @@ public class CelebornTezShuffleDataFetcher extends CallableWithNdc<Void> {
return partitionId;
}
- public void setPartitionId(Integer partitionId) {
- this.partitionId = partitionId;
- }
-
private boolean issueMapOutputMerge() throws IOException {
// Allocate a MapOutput (either in-memory or on-disk) to put uncompressed block
// In Rss, a MapOutput is sent as multiple blocks, so the reducer needs to
@@ -245,12 +238,12 @@ public class CelebornTezShuffleDataFetcher extends CallableWithNdc<Void> {
}
private void stopFetch() {
- LOG.info("RssTezShuffleDataFetcher stop fetch");
+ LOG.info("CelebornTezShuffleDataFetcher stop fetch");
stopped = true;
}
public void shutDown() {
stopFetch();
- LOG.info("RssTezShuffleDataFetcher shutdown");
+ LOG.info("CelebornTezShuffleDataFetcher shutdown");
}
}
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/writers/CelebornUnorderedPartitionedKVWriter.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/writers/CelebornUnorderedPartitionedKVWriter.java
index 492a2cacb..c86fde957 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/writers/CelebornUnorderedPartitionedKVWriter.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/common/writers/CelebornUnorderedPartitionedKVWriter.java
@@ -18,15 +18,10 @@
package org.apache.tez.runtime.library.common.writers;
import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.BitSet;
import java.util.Iterator;
-import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.zip.Deflater;
-import com.google.protobuf.ByteString;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.RawLocalFileSystem;
@@ -34,21 +29,15 @@ import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.serializer.Serialization;
import org.apache.hadoop.io.serializer.SerializationFactory;
import org.apache.hadoop.io.serializer.Serializer;
-import org.apache.tez.common.Preconditions;
import org.apache.tez.common.TezCommonUtils;
-import org.apache.tez.common.TezUtilsInternal;
import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.common.counters.TezCounter;
-import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.OutputContext;
-import org.apache.tez.runtime.api.events.CompositeDataMovementEvent;
import org.apache.tez.runtime.library.api.KeyValuesWriter;
import org.apache.tez.runtime.library.api.Partitioner;
import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.ConfigUtils;
import org.apache.tez.runtime.library.common.TezRuntimeUtils;
-import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
-import org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads;
import org.apache.tez.runtime.library.utils.CodecUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -201,76 +190,10 @@ public class CelebornUnorderedPartitionedKVWriter extends KeyValuesWriter {
pusher.insert(key, value, partitioner.getPartition(key, value, numOutputs));
}
- public List<Event> close() throws IOException {
+ public void close() throws IOException {
pusher.close();
isShutdown.set(true);
updateTezCountersAndNotify();
- List<Event> eventList = new ArrayList<>();
- eventList.add(generateVMEvent());
- eventList.add(generateDMEvent());
- return eventList;
- }
-
- private Event generateVMEvent() throws IOException {
- return ShuffleUtils.generateVMEvent(
- outputContext, this.sizePerPartition, reportPartitionStats.isPrecise(), deflater.get());
- }
-
- private Event generateDMEvent() throws IOException {
- BitSet emptyPartitions = getEmptyPartitions(numRecordsPerPartition);
- return generateDMEvent(false, -1, false, outputContext.getUniqueIdentifier(), emptyPartitions);
- }
-
- private Event generateDMEvent(
- boolean addSpillDetails,
- int spillId,
- boolean isLastSpill,
- String pathComponent,
- BitSet emptyPartitions)
- throws IOException {
-
- outputContext.notifyProgress();
- ShuffleUserPayloads.DataMovementEventPayloadProto.Builder payloadBuilder =
- ShuffleUserPayloads.DataMovementEventPayloadProto.newBuilder();
- if (numOutputs == 1) {
- payloadBuilder.setNumRecord((int) outputRecordsCounter.getValue());
- }
-
- String host = "fake";
- if (emptyPartitions.cardinality() != 0) {
- // Empty partitions exist
- ByteString emptyPartitionsByteString =
- TezCommonUtils.compressByteArrayToByteString(
- TezUtilsInternal.toByteArray(emptyPartitions), deflater.get());
- payloadBuilder.setEmptyPartitions(emptyPartitionsByteString);
- }
-
- if (emptyPartitions.cardinality() != numOutputs) {
- // Populate payload only if at least 1 partition has data
- payloadBuilder.setHost(host);
- payloadBuilder.setPort(0);
- payloadBuilder.setPathComponent(pathComponent);
- }
-
- if (addSpillDetails) {
- payloadBuilder.setSpillId(spillId);
- payloadBuilder.setLastEvent(isLastSpill);
- }
-
- ByteBuffer payload = payloadBuilder.build().toByteString().asReadOnlyByteBuffer();
- return CompositeDataMovementEvent.create(0, numOutputs, payload);
- }
-
- private BitSet getEmptyPartitions(int[] recordsPerPartition) {
- Preconditions.checkArgument(
- recordsPerPartition != null, "records per partition can not be null");
- BitSet emptyPartitions = new BitSet();
- for (int i = 0; i < numOutputs; i++) {
- if (recordsPerPartition[i] == 0) {
- emptyPartitions.set(i);
- }
- }
- return emptyPartitions;
}
private void updateTezCountersAndNotify() {
@@ -283,4 +206,16 @@ public class CelebornUnorderedPartitionedKVWriter extends KeyValuesWriter {
}
outputContext.notifyProgress();
}
+
+ public int[] getNumRecordsPerPartition() {
+ return numRecordsPerPartition;
+ }
+
+ public boolean reportDetailedPartitionStats() {
+ return reportPartitionStats.isPrecise();
+ }
+
+ public long[] getPartitionStats() {
+ return sizePerPartition;
+ }
}
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/input/CelebornOrderedGroupedKVInput.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/input/CelebornOrderedGroupedKVInput.java
index 7790001a4..627d5f7a7 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/input/CelebornOrderedGroupedKVInput.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/input/CelebornOrderedGroupedKVInput.java
@@ -16,16 +16,13 @@ package org.apache.tez.runtime.library.input;
import java.io.IOException;
import java.util.Collections;
-import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
-import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.annotations.VisibleForTesting;
-import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.RawComparator;
@@ -35,7 +32,6 @@ import org.apache.tez.common.TezRuntimeFrameworkConfigs;
import org.apache.tez.common.TezUtils;
import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.common.counters.TezCounter;
-import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezException;
import org.apache.tez.runtime.api.AbstractLogicalInput;
import org.apache.tez.runtime.api.Event;
@@ -43,9 +39,7 @@ import org.apache.tez.runtime.api.InputContext;
import org.apache.tez.runtime.api.ProgressFailedException;
import org.apache.tez.runtime.library.api.IOInterruptedException;
import org.apache.tez.runtime.library.api.KeyValuesReader;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.ConfigUtils;
-import org.apache.tez.runtime.library.common.Constants;
import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
import org.apache.tez.runtime.library.common.ValuesIterator;
import org.apache.tez.runtime.library.common.shuffle.orderedgrouped.CelebornShuffle;
@@ -73,7 +67,7 @@ public class CelebornOrderedGroupedKVInput extends AbstractLogicalInput {
protected TezRawKeyValueIterator rawIter = null;
protected Configuration conf;
- protected CelebornShuffle shuffle;
+ protected Shuffle shuffle;
private ApplicationAttemptId applicationAttemptId;
protected MemoryUpdateCallbackHandler memoryUpdateCallbackHandler;
private final BlockingQueue<Event> pendingEvents = new LinkedBlockingQueue<Event>();
@@ -145,7 +139,7 @@ public class CelebornOrderedGroupedKVInput extends AbstractLogicalInput {
}
@VisibleForTesting
- CelebornShuffle createShuffle() throws IOException {
+ Shuffle createShuffle() throws IOException {
return new CelebornShuffle(
getContext(),
conf,
@@ -179,7 +173,7 @@ public class CelebornOrderedGroupedKVInput extends AbstractLogicalInput {
*/
public void waitForInputReady() throws IOException, InterruptedException, TezException {
// Cannot synchronize entire method since this is called form user code and can block.
- CelebornShuffle localShuffleCopy = null;
+ Shuffle localShuffleCopy = null;
synchronized (this) {
Preconditions.checkState(isStarted.get(), "Must start input before invoking this method");
if (getNumPhysicalInputs() == 0) {
@@ -285,7 +279,7 @@ public class CelebornOrderedGroupedKVInput extends AbstractLogicalInput {
@Override
public void handleEvents(List<Event> inputEvents) throws IOException {
- CelebornShuffle shuffleLocalRef;
+ Shuffle shuffleLocalRef;
synchronized (this) {
if (getNumPhysicalInputs() == 0) {
throw new RuntimeException("No input events expected as numInputs is 0");
@@ -356,68 +350,5 @@ public class CelebornOrderedGroupedKVInput extends AbstractLogicalInput {
public Iterable<Object> getCurrentValues() throws IOException {
return valuesIter.getValues();
}
- };
-
- private static final Set<String> confKeys = new HashSet<String>();
-
- static {
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINE_MIN_SPILLS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_VERIFY_DISK_CHECKSUM);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_BUFFER_PERCENT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMTOMEM_SEGMENTS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_MEMTOMEM);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_SOURCE_ATTEMPT_ABORT_LIMIT);
- confKeys.add(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ACCEPTABLE_HOST_FETCH_FAILURE_FRACTION);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MIN_FAILURES_PER_HOST);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MAX_STALL_TIME_FRACTION);
- confKeys.add(
- TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MAX_ALLOWED_FAILED_FETCH_ATTEMPT_FRACTION);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MIN_REQUIRED_PROGRESS_FRACTION);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FAILED_CHECK_SINCE_LAST_COMPLETION);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCHER_USE_SHARED_POOL);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_POST_MERGE_BUFFER_PERCENT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_GROUP_COMPARATOR_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_SECONDARY_COMPARATOR_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CONVERT_USER_PAYLOAD_TO_HISTORY_TEXT);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_GROUP_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_COUNTER_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX_GROUPS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CLEANUP_FILES_ON_INTERRUPT);
- confKeys.add(Constants.TEZ_RUNTIME_TASK_MEMORY);
- confKeys.add(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID);
- }
-
- // TODO Maybe add helper methods to extract keys
- // TODO Maybe add constants or an Enum to access the keys
-
- @InterfaceAudience.Private
- public static Set<String> getConfigurationKeySet() {
- return Collections.unmodifiableSet(confKeys);
}
}
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/input/CelebornUnorderedKVInput.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/input/CelebornUnorderedKVInput.java
index 1d29f807d..4a1079709 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/input/CelebornUnorderedKVInput.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/input/CelebornUnorderedKVInput.java
@@ -18,15 +18,12 @@ import static org.apache.celeborn.tez.plugin.util.CelebornTezUtils.*;
import java.io.IOException;
import java.util.Collections;
-import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
-import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.compress.CompressionCodec;
@@ -37,14 +34,12 @@ import org.apache.tez.common.TezUtils;
import org.apache.tez.common.TezUtilsInternal;
import org.apache.tez.common.counters.TaskCounter;
import org.apache.tez.common.counters.TezCounter;
-import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.runtime.api.AbstractLogicalInput;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.InputContext;
import org.apache.tez.runtime.api.ProgressFailedException;
import org.apache.tez.runtime.library.api.KeyValueReader;
import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.apache.tez.runtime.library.common.Constants;
import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
import org.apache.tez.runtime.library.common.readers.UnorderedKVReader;
import org.apache.tez.runtime.library.common.shuffle.ShuffleEventHandler;
@@ -159,11 +154,17 @@ public class CelebornUnorderedKVInput extends AbstractLogicalInput {
int port = conf.getInt(TEZ_CELEBORN_LM_PORT, -1);
int shuffleId = conf.getInt(TEZ_SHUFFLE_ID, -1);
String appId = conf.get(TEZ_CELEBORN_APPLICATION_ID);
- String user = conf.get(TEZ_CELEBORN_USER);
boolean broadcastOrOneToOne = conf.getBoolean(TEZ_BROADCAST_OR_ONETOONE, false);
CelebornConf celebornConf = CelebornTezUtils.fromTezConfiguration(conf);
ShuffleClient shuffleClient =
- ShuffleClient.get(appId, host, port, celebornConf, UserIdentifier.apply(user), null);
+ ShuffleClient.get(
+ appId,
+ host,
+ port,
+ celebornConf,
+ new UserIdentifier(
+ celebornConf.quotaUserSpecificTenant(), celebornConf.quotaUserSpecificUserName()),
+ null);
this.shuffleManager =
new CelebornShuffleManager(
@@ -305,54 +306,6 @@ public class CelebornUnorderedKVInput extends AbstractLogicalInput {
getContext());
}
- private static final Set<String> confKeys = new HashSet<String>();
-
- static {
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_ENABLE_SSL);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_VERIFY_DISK_CHECKSUM);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCH_BUFFER_PERCENT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_FETCHER_USE_SHARED_POOL);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_POST_MERGE_BUFFER_PERCENT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_OPTIMIZE_SHARED_FETCH);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CONVERT_USER_PAYLOAD_TO_HISTORY_TEXT);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_GROUP_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_COUNTER_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX_GROUPS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CLEANUP_FILES_ON_INTERRUPT);
- confKeys.add(Constants.TEZ_RUNTIME_TASK_MEMORY);
- confKeys.add(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID);
- }
-
- // TODO Maybe add helper methods to extract keys
- // TODO Maybe add constants or an Enum to access the keys
-
- @InterfaceAudience.Private
- public static Set<String> getConfigurationKeySet() {
- return Collections.unmodifiableSet(confKeys);
- }
-
@Override
public float getProgress() throws ProgressFailedException, InterruptedException {
try {
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornOrderedPartitionedKVOutput.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornOrderedPartitionedKVOutput.java
index 4577337e0..0c379f410 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornOrderedPartitionedKVOutput.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornOrderedPartitionedKVOutput.java
@@ -17,38 +17,20 @@ package org.apache.tez.runtime.library.output;
import static org.apache.celeborn.tez.plugin.util.CelebornTezUtils.*;
import java.io.IOException;
-import java.util.Arrays;
import java.util.Collections;
-import java.util.HashSet;
import java.util.List;
-import java.util.Locale;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.zip.Deflater;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
-import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceAudience.Public;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.RawLocalFileSystem;
-import org.apache.tez.common.Preconditions;
-import org.apache.tez.common.TezCommonUtils;
-import org.apache.tez.common.TezRuntimeFrameworkConfigs;
-import org.apache.tez.common.TezUtils;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.records.TezTaskAttemptID;
import org.apache.tez.runtime.api.AbstractLogicalOutput;
import org.apache.tez.runtime.api.Event;
import org.apache.tez.runtime.api.OutputContext;
-import org.apache.tez.runtime.library.api.KeyValuesWriter;
import org.apache.tez.runtime.library.api.Partitioner;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
-import org.apache.tez.runtime.library.common.sort.impl.ExternalSorter;
-import org.apache.tez.runtime.library.conf.OrderedPartitionedKVOutputConfig.SorterImpl;
import org.apache.tez.runtime.library.sort.CelebornSorter;
import org.apache.tez.runtime.library.sort.CelebornTezPerPartitionRecord;
import org.slf4j.Logger;
@@ -64,16 +46,11 @@ import org.apache.celeborn.tez.plugin.util.CelebornTezUtils;
* key/value pairs written to it. It also partitions the output based on a {@link Partitioner}
*/
@Public
-public class CelebornOrderedPartitionedKVOutput extends AbstractLogicalOutput {
+public class CelebornOrderedPartitionedKVOutput extends OrderedPartitionedKVOutput {
private static final Logger LOG =
LoggerFactory.getLogger(CelebornOrderedPartitionedKVOutput.class);
- protected ExternalSorter sorter;
- protected Configuration conf;
- private RawLocalFileSystem localFs;
- protected MemoryUpdateCallbackHandler memoryUpdateCallbackHandler;
- private long startTime;
private long endTime;
private int mapNum;
private int numOutputs;
@@ -86,14 +63,14 @@ public class CelebornOrderedPartitionedKVOutput extends AbstractLogicalOutput {
private final AtomicBoolean isStarted = new AtomicBoolean(false);
private final Deflater deflater;
CelebornTezWriter celebornTezWriter;
- CelebornConf celebornConf;
- @VisibleForTesting boolean pipelinedShuffle;
private boolean sendEmptyPartitionDetails;
- @VisibleForTesting boolean finalMergeEnabled;
+ CelebornConf celebornConf;
public CelebornOrderedPartitionedKVOutput(OutputContext outputContext, int numPhysicalOutputs) {
super(outputContext, numPhysicalOutputs);
- deflater = TezCommonUtils.newBestCompressionDeflater();
+ this.deflater = (Deflater) getParentPrivateField(this, "deflater");
+ this.sendEmptyPartitionDetails =
+ (boolean) getParentPrivateField(this, "sendEmptyPartitionDetails");
this.numOutputs = getNumPhysicalOutputs();
this.mapNum = outputContext.getVertexParallelism();
TezTaskAttemptID taskAttemptId =
@@ -105,31 +82,12 @@ public class CelebornOrderedPartitionedKVOutput extends AbstractLogicalOutput {
@Override
public synchronized List<Event> initialize() throws IOException {
- this.startTime = System.nanoTime();
- this.conf = TezUtils.createConfFromBaseConfAndPayload(getContext());
- this.localFs = (RawLocalFileSystem) FileSystem.getLocal(conf).getRaw();
-
- // Initializing this parametr in this conf since it is used in multiple
- // places (wherever LocalDirAllocator is used) - TezTaskOutputFiles,
- // TezMerger, etc.
- this.conf.setStrings(TezRuntimeFrameworkConfigs.LOCAL_DIRS, getContext().getWorkDirs());
- this.memoryUpdateCallbackHandler = new MemoryUpdateCallbackHandler();
- getContext()
- .requestInitialMemory(
- ExternalSorter.getInitialMemoryRequirement(
- conf, getContext().getTotalMemoryAvailableToTask()),
- memoryUpdateCallbackHandler);
-
- sendEmptyPartitionDetails =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED,
- TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED_DEFAULT);
+ super.initialize();
this.host = this.conf.get(TEZ_CELEBORN_LM_HOST);
this.port = this.conf.getInt(TEZ_CELEBORN_LM_PORT, -1);
this.shuffleId = this.conf.getInt(TEZ_SHUFFLE_ID, -1);
this.appId = this.conf.get(TEZ_CELEBORN_APPLICATION_ID);
- String user = this.conf.get(TEZ_CELEBORN_USER);
celebornConf = CelebornTezUtils.fromTezConfiguration(conf);
celebornTezWriter =
new CelebornTezWriter(
@@ -143,95 +101,24 @@ public class CelebornOrderedPartitionedKVOutput extends AbstractLogicalOutput {
appId,
host,
port,
- UserIdentifier.apply(user));
+ new UserIdentifier(
+ celebornConf.quotaUserSpecificTenant(), celebornConf.quotaUserSpecificUserName()));
return Collections.emptyList();
}
@Override
public synchronized void start() throws Exception {
- if (!isStarted.get()) {
- memoryUpdateCallbackHandler.validateUpdateReceived();
- String sorterClass =
- conf.get(
- TezRuntimeConfiguration.TEZ_RUNTIME_SORTER_CLASS,
- TezRuntimeConfiguration.TEZ_RUNTIME_SORTER_CLASS_DEFAULT)
- .toUpperCase(Locale.ENGLISH);
- SorterImpl sorterImpl = null;
- try {
- sorterImpl = SorterImpl.valueOf(sorterClass);
- } catch (IllegalArgumentException e) {
- throw new IllegalArgumentException(
- "Invalid sorter class specified in config"
- + ", propertyName="
- + TezRuntimeConfiguration.TEZ_RUNTIME_SORTER_CLASS
- + ", value="
- + sorterClass
- + ", validValues="
- + Arrays.asList(SorterImpl.values()));
- }
-
- finalMergeEnabled =
- conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT,
- TezRuntimeConfiguration.TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT_DEFAULT);
-
- pipelinedShuffle =
- this.conf.getBoolean(
- TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED,
- TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED_DEFAULT);
-
- if (pipelinedShuffle) {
- if (finalMergeEnabled) {
- LOG.info(
- getContext().getInputOutputVertexNames()
- + " 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);
- }
-
- Preconditions.checkArgument(
- sorterImpl.equals(SorterImpl.PIPELINED),
- TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED
- + "only works with PipelinedSorter.");
- }
-
- sorter =
- new CelebornSorter(
- getContext(),
- conf,
- getNumPhysicalOutputs(),
- (int) memoryUpdateCallbackHandler.getMemoryAssigned(),
- celebornTezWriter,
- celebornConf);
-
- isStarted.set(true);
- }
- }
-
- @Override
- public synchronized KeyValuesWriter getWriter() throws IOException {
- com.google.common.base.Preconditions.checkState(
- isStarted.get(), "Cannot get writer before starting the Output");
-
- return new KeyValuesWriter() {
- @Override
- public void write(Object key, Iterable<Object> values) throws IOException {
- sorter.write(key, values);
- }
-
- @Override
- public void write(Object key, Object value) throws IOException {
- sorter.write(key, value);
- }
- };
- }
-
- @Override
- public synchronized void handleEvents(List<Event> outputEvents) {
- // Not expecting any events.
+ super.start();
+ sorter =
+ new CelebornSorter(
+ getContext(),
+ conf,
+ getNumPhysicalOutputs(),
+ (int) memoryUpdateCallbackHandler.getMemoryAssigned(),
+ celebornTezWriter,
+ celebornConf);
+ setParentPrivateField(this, "sorter", sorter);
}
@Override
@@ -244,12 +131,7 @@ public class CelebornOrderedPartitionedKVOutput extends AbstractLogicalOutput {
returnEvents.addAll(generateEvents());
sorter = null;
} else {
- LOG.warn(
- getContext().getInputOutputVertexNames()
- + ": Attempting to close output {} of type {} before it was started. Generating empty events",
- getContext().getDestinationVertexName(),
- this.getClass().getSimpleName());
- returnEvents = generateEmptyEvents();
+ returnEvents = super.close();
}
return returnEvents;
@@ -287,55 +169,4 @@ public class CelebornOrderedPartitionedKVOutput extends AbstractLogicalOutput {
LOG.info("Generate events.");
return eventList;
}
-
- private List<Event> generateEmptyEvents() throws IOException {
- List<Event> eventList = Lists.newLinkedList();
- ShuffleUtils.generateEventsForNonStartedOutput(
- eventList, getNumPhysicalOutputs(), getContext(), true, true, deflater);
- return eventList;
- }
-
- private static final Set<String> confKeys = new HashSet<String>();
-
- static {
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SORT_SPILL_PERCENT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_REPORT_PARTITION_STATS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINE_MIN_SPILLS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SORTER_SORT_THREADS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SORTER_MIN_BLOCK_SIZE_IN_MB);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SORTER_LAZY_ALLOCATE_MEMORY);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CONVERT_USER_PAYLOAD_TO_HISTORY_TEXT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_GROUP_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_COUNTER_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX_GROUPS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_SORTER_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CLEANUP_FILES_ON_INTERRUPT);
- confKeys.add(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID);
- }
-
- // TODO Maybe add helper methods to extract keys
- // TODO Maybe add constants or an Enum to access the keys
-
- @InterfaceAudience.Private
- public static Set<String> getConfigurationKeySet() {
- return Collections.unmodifiableSet(confKeys);
- }
}
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornUnorderedKVOutput.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornUnorderedKVOutput.java
index f30a16440..77300e8df 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornUnorderedKVOutput.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornUnorderedKVOutput.java
@@ -16,14 +16,16 @@ package org.apache.tez.runtime.library.output;
import static org.apache.celeborn.tez.plugin.util.CelebornTezUtils.*;
+import java.io.IOException;
+import java.util.BitSet;
import java.util.Collections;
-import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.zip.Deflater;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.conf.Configuration;
@@ -43,8 +45,10 @@ import org.apache.tez.runtime.library.api.Partitioner;
import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
+import org.apache.tez.runtime.library.common.sort.impl.TezIndexRecord;
import org.apache.tez.runtime.library.common.writers.CelebornUnorderedPartitionedKVWriter;
import org.apache.tez.runtime.library.common.writers.UnorderedPartitionedKVWriter;
+import org.apache.tez.runtime.library.sort.CelebornTezPerPartitionRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,6 +71,8 @@ public class CelebornUnorderedKVOutput extends AbstractLogicalOutput {
@VisibleForTesting Configuration conf;
private MemoryUpdateCallbackHandler memoryUpdateCallbackHandler;
private CelebornUnorderedPartitionedKVWriter kvWriter;
+ private final Deflater deflater;
+ private boolean sendEmptyPartitionDetails;
private final AtomicBoolean isStarted = new AtomicBoolean(false);
private static int mapId;
@@ -88,6 +94,7 @@ public class CelebornUnorderedKVOutput extends AbstractLogicalOutput {
CelebornTezUtils.uniqueIdentifierToAttemptId(outputContext.getUniqueIdentifier()));
attemptId = taskAttemptId.getId();
mapId = taskAttemptId.getTaskID().getId();
+ deflater = TezCommonUtils.newBestCompressionDeflater();
}
@Override
@@ -98,6 +105,10 @@ public class CelebornUnorderedKVOutput extends AbstractLogicalOutput {
TezRuntimeFrameworkConfigs.TEZ_RUNTIME_NUM_EXPECTED_PARTITIONS, getNumPhysicalOutputs());
this.conf.set(
TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, CustomPartitioner.class.getName());
+ sendEmptyPartitionDetails =
+ conf.getBoolean(
+ TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED,
+ TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED_DEFAULT);
this.memoryUpdateCallbackHandler = new MemoryUpdateCallbackHandler();
getContext()
.requestInitialMemory(
@@ -117,9 +128,15 @@ public class CelebornUnorderedKVOutput extends AbstractLogicalOutput {
if (!isStarted.get()) {
memoryUpdateCallbackHandler.validateUpdateReceived();
CelebornConf celebornConf = CelebornTezUtils.fromTezConfiguration(conf);
- String user = this.conf.get(TEZ_CELEBORN_USER);
ShuffleClient shuffleClient =
- ShuffleClient.get(appId, host, port, celebornConf, UserIdentifier.apply(user));
+ ShuffleClient.get(
+ appId,
+ host,
+ port,
+ celebornConf,
+ new UserIdentifier(
+ celebornConf.quotaUserSpecificTenant(),
+ celebornConf.quotaUserSpecificUserName()));
this.kvWriter =
new CelebornUnorderedPartitionedKVWriter(
getContext(),
@@ -148,9 +165,10 @@ public class CelebornUnorderedKVOutput extends AbstractLogicalOutput {
@Override
public synchronized List<Event> close() throws Exception {
- List<Event> returnEvents = null;
+ List<Event> returnEvents;
if (isStarted.get()) {
- returnEvents = kvWriter.close();
+ kvWriter.close();
+ returnEvents = generateEvents();
kvWriter = null;
} else {
LOG.warn(
@@ -179,44 +197,47 @@ public class CelebornUnorderedKVOutput extends AbstractLogicalOutput {
return returnEvents;
}
- private static final Set<String> confKeys = new HashSet<String>();
-
- static {
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_MAX_SIZE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_SUPPORT_IN_MEM_FILE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CONVERT_USER_PAYLOAD_TO_HISTORY_TEXT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_GROUP_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_COUNTER_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX_GROUPS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CLEANUP_FILES_ON_INTERRUPT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_REPORT_PARTITION_STATS);
- confKeys.add(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID);
- confKeys.add(
- TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_PARTITIONED_KVWRITER_BUFFER_MERGE_PERCENT);
- }
+ private List<Event> generateEvents() throws IOException {
+ List<Event> eventList = Lists.newLinkedList();
+ boolean isLastEvent = true;
- // TODO Maybe add helper methods to extract keys
- // TODO Maybe add constants or an Enum to access the keys
+ String auxiliaryService =
+ conf.get(
+ TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID,
+ TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT);
- @InterfaceAudience.Private
- public static Set<String> getConfigurationKeySet() {
- return Collections.unmodifiableSet(confKeys);
+ int[] numRecordsPerPartition = kvWriter.getNumRecordsPerPartition();
+
+ CelebornTezPerPartitionRecord celebornTezPerPartitionRecord =
+ new CelebornTezPerPartitionRecord(numOutputs, numRecordsPerPartition);
+
+ BitSet emptyPartitionDetails = new BitSet();
+ for (int i = 0; i < celebornTezPerPartitionRecord.size(); i++) {
+ TezIndexRecord indexRecord = celebornTezPerPartitionRecord.getIndex(i);
+ if (!indexRecord.hasData()) {
+ emptyPartitionDetails.set(i);
+ }
+ }
+ if (emptyPartitionDetails.cardinality() > 0) {
+ LOG.info("empty partition details");
+ }
+
+ ShuffleUtils.generateEventOnSpill(
+ eventList,
+ true,
+ isLastEvent,
+ getContext(),
+ 0,
+ celebornTezPerPartitionRecord,
+ getNumPhysicalOutputs(),
+ sendEmptyPartitionDetails,
+ getContext().getUniqueIdentifier(),
+ kvWriter.getPartitionStats(),
+ kvWriter.reportDetailedPartitionStats(),
+ auxiliaryService,
+ deflater);
+ LOG.info("Generate events.");
+ return eventList;
}
@InterfaceAudience.Private
diff --git a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornUnorderedPartitionedKVOutput.java b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornUnorderedPartitionedKVOutput.java
index 98b82ad3b..44d94f873 100644
--- a/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornUnorderedPartitionedKVOutput.java
+++ b/client-tez/tez/src/main/java/org/apache/tez/runtime/library/output/CelebornUnorderedPartitionedKVOutput.java
@@ -16,15 +16,15 @@ package org.apache.tez.runtime.library.output;
import static org.apache.celeborn.tez.plugin.util.CelebornTezUtils.*;
+import java.io.IOException;
import java.util.Collections;
-import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
-import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.zip.Deflater;
import com.google.common.annotations.VisibleForTesting;
-import org.apache.hadoop.classification.InterfaceAudience;
+import com.google.common.collect.Lists;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.conf.Configuration;
import org.apache.tez.common.Preconditions;
@@ -44,6 +44,7 @@ import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
import org.apache.tez.runtime.library.common.shuffle.ShuffleUtils;
import org.apache.tez.runtime.library.common.writers.CelebornUnorderedPartitionedKVWriter;
import org.apache.tez.runtime.library.common.writers.UnorderedPartitionedKVWriter;
+import org.apache.tez.runtime.library.sort.CelebornTezPerPartitionRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -66,6 +67,9 @@ public class CelebornUnorderedPartitionedKVOutput extends AbstractLogicalOutput
@VisibleForTesting Configuration conf;
private MemoryUpdateCallbackHandler memoryUpdateCallbackHandler;
private CelebornUnorderedPartitionedKVWriter kvWriter;
+ private final Deflater deflater;
+
+ boolean sendEmptyPartitionDetails;
private final AtomicBoolean isStarted = new AtomicBoolean(false);
private static int mapId;
@@ -86,6 +90,7 @@ public class CelebornUnorderedPartitionedKVOutput extends AbstractLogicalOutput
CelebornTezUtils.uniqueIdentifierToAttemptId(outputContext.getUniqueIdentifier()));
attemptId = taskAttemptId.getId();
mapId = taskAttemptId.getTaskID().getId();
+ deflater = TezCommonUtils.newBestCompressionDeflater();
}
@Override
@@ -94,6 +99,10 @@ public class CelebornUnorderedPartitionedKVOutput extends AbstractLogicalOutput
this.conf.setStrings(TezRuntimeFrameworkConfigs.LOCAL_DIRS, getContext().getWorkDirs());
this.conf.setInt(
TezRuntimeFrameworkConfigs.TEZ_RUNTIME_NUM_EXPECTED_PARTITIONS, getNumPhysicalOutputs());
+ sendEmptyPartitionDetails =
+ conf.getBoolean(
+ TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED,
+ TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED_DEFAULT);
this.memoryUpdateCallbackHandler = new MemoryUpdateCallbackHandler();
getContext()
.requestInitialMemory(
@@ -113,9 +122,15 @@ public class CelebornUnorderedPartitionedKVOutput extends AbstractLogicalOutput
if (!isStarted.get()) {
memoryUpdateCallbackHandler.validateUpdateReceived();
CelebornConf celebornConf = CelebornTezUtils.fromTezConfiguration(conf);
- String user = this.conf.get(TEZ_CELEBORN_USER);
ShuffleClient shuffleClient =
- ShuffleClient.get(appId, host, port, celebornConf, UserIdentifier.apply(user));
+ ShuffleClient.get(
+ appId,
+ host,
+ port,
+ celebornConf,
+ new UserIdentifier(
+ celebornConf.quotaUserSpecificTenant(),
+ celebornConf.quotaUserSpecificUserName()));
this.kvWriter =
new CelebornUnorderedPartitionedKVWriter(
getContext(),
@@ -144,9 +159,10 @@ public class CelebornUnorderedPartitionedKVOutput extends AbstractLogicalOutput
@Override
public synchronized List<Event> close() throws Exception {
- List<Event> returnEvents = null;
+ List<Event> returnEvents;
if (isStarted.get()) {
- returnEvents = kvWriter.close();
+ kvWriter.close();
+ returnEvents = generateEvents();
kvWriter = null;
} else {
LOG.warn(
@@ -175,43 +191,35 @@ public class CelebornUnorderedPartitionedKVOutput extends AbstractLogicalOutput
return returnEvents;
}
- private static final Set<String> confKeys = new HashSet<String>();
-
- static {
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_MAX_SIZE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_SUPPORT_IN_MEM_FILE);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CONVERT_USER_PAYLOAD_TO_HISTORY_TEXT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_GROUP_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_COUNTER_NAME_MAX_LENGTH);
- confKeys.add(TezConfiguration.TEZ_COUNTERS_MAX_GROUPS);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_CLEANUP_FILES_ON_INTERRUPT);
- confKeys.add(TezRuntimeConfiguration.TEZ_RUNTIME_REPORT_PARTITION_STATS);
- confKeys.add(TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID);
- confKeys.add(
- TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_PARTITIONED_KVWRITER_BUFFER_MERGE_PERCENT);
- }
-
- // TODO Maybe add helper methods to extract keys
- // TODO Maybe add constants or an Enum to access the keys
-
- @InterfaceAudience.Private
- public static Set<String> getConfigurationKeySet() {
- return Collections.unmodifiableSet(confKeys);
+ private List<Event> generateEvents() throws IOException {
+ List<Event> eventList = Lists.newLinkedList();
+ boolean isLastEvent = true;
+
+ String auxiliaryService =
+ conf.get(
+ TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID,
+ TezConfiguration.TEZ_AM_SHUFFLE_AUXILIARY_SERVICE_ID_DEFAULT);
+
+ int[] numRecordsPerPartition = kvWriter.getNumRecordsPerPartition();
+
+ CelebornTezPerPartitionRecord celebornTezPerPartitionRecord =
+ new CelebornTezPerPartitionRecord(numOutputs, numRecordsPerPartition);
+
+ ShuffleUtils.generateEventOnSpill(
+ eventList,
+ true,
+ isLastEvent,
+ getContext(),
+ 0,
+ celebornTezPerPartitionRecord,
+ getNumPhysicalOutputs(),
+ sendEmptyPartitionDetails,
+ getContext().getUniqueIdentifier(),
+ kvWriter.getPartitionStats(),
+ kvWriter.reportDetailedPartitionStats(),
+ auxiliaryService,
+ deflater);
+ LOG.info("Generate events.");
+ return eventList;
}
}
diff --git a/client-tez/tez/src/test/java/org/apache/tez/runtime/library/output/OutputTestHelpers.java b/client-tez/tez/src/test/java/org/apache/tez/runtime/library/output/OutputTestHelpers.java
new file mode 100644
index 000000000..0b37fe0c8
--- /dev/null
+++ b/client-tez/tez/src/test/java/org/apache/tez/runtime/library/output/OutputTestHelpers.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tez.runtime.library.output;
+
+import static org.mockito.Mockito.*;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.UserPayload;
+import org.apache.tez.runtime.api.OutputContext;
+import org.apache.tez.runtime.api.OutputStatisticsReporter;
+import org.apache.tez.runtime.api.impl.ExecutionContextImpl;
+import org.apache.tez.runtime.library.common.MemoryUpdateCallbackHandler;
+import org.mockito.ArgumentMatchers;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+final class OutputTestHelpers {
+
+ private OutputTestHelpers() {}
+
+ static OutputContext createOutputContext() throws IOException {
+ OutputContext outputContext = mock(OutputContext.class);
+ Configuration conf = new TezConfiguration();
+ UserPayload payLoad = TezUtils.createUserPayloadFromConf(conf);
+ String[] workingDirs = new String[] {"workDir1"};
+ OutputStatisticsReporter statsReporter = mock(OutputStatisticsReporter.class);
+ TezCounters counters = new TezCounters();
+
+ doReturn("destinationVertex").when(outputContext).getDestinationVertexName();
+ doReturn(payLoad).when(outputContext).getUserPayload();
+ doReturn(workingDirs).when(outputContext).getWorkDirs();
+ doReturn(200 * 1024 * 1024L).when(outputContext).getTotalMemoryAvailableToTask();
+ doReturn(counters).when(outputContext).getCounters();
+ doReturn(statsReporter).when(outputContext).getStatisticsReporter();
+ doReturn(new Configuration(false)).when(outputContext).getContainerConfiguration();
+ return outputContext;
+ }
+
+ static OutputContext createOutputContext(
+ Configuration conf, Configuration userPayloadConf, Path workingDir) throws IOException {
+ OutputContext ctx = mock(OutputContext.class);
+ doAnswer(
+ new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) {
+ long requestedSize = (Long) invocation.getArguments()[0];
+ MemoryUpdateCallbackHandler callback =
+ (MemoryUpdateCallbackHandler) invocation.getArguments()[1];
+ callback.memoryAssigned(requestedSize);
+ return null;
+ }
+ })
+ .when(ctx)
+ .requestInitialMemory(ArgumentMatchers.anyLong(), ArgumentMatchers.any());
+ doReturn(conf).when(ctx).getContainerConfiguration();
+ doReturn(TezUtils.createUserPayloadFromConf(userPayloadConf)).when(ctx).getUserPayload();
+ doReturn("taskVertex").when(ctx).getTaskVertexName();
+ doReturn("destinationVertex").when(ctx).getDestinationVertexName();
+ doReturn("attempt_1681717153064_3601637_1_13_000096_0").when(ctx).getUniqueIdentifier();
+ doReturn(new String[] {workingDir.toString()}).when(ctx).getWorkDirs();
+ doReturn(200 * 1024 * 1024L).when(ctx).getTotalMemoryAvailableToTask();
+ doReturn(new TezCounters()).when(ctx).getCounters();
+ OutputStatisticsReporter statsReporter = mock(OutputStatisticsReporter.class);
+ doReturn(statsReporter).when(ctx).getStatisticsReporter();
+ doReturn(new ExecutionContextImpl("localhost")).when(ctx).getExecutionContext();
+ return ctx;
+ }
+}
diff --git a/client-tez/tez/src/test/java/org/apache/tez/runtime/library/output/TestOrderedPartitionedKVOutput.java b/client-tez/tez/src/test/java/org/apache/tez/runtime/library/output/TestOrderedPartitionedKVOutput.java
new file mode 100644
index 000000000..d4e33e87a
--- /dev/null
+++ b/client-tez/tez/src/test/java/org/apache/tez/runtime/library/output/TestOrderedPartitionedKVOutput.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tez.runtime.library.output;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.BitSet;
+import java.util.List;
+
+import com.google.protobuf.ByteString;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.tez.common.TezCommonUtils;
+import org.apache.tez.common.TezRuntimeFrameworkConfigs;
+import org.apache.tez.common.TezUtilsInternal;
+import org.apache.tez.runtime.api.Event;
+import org.apache.tez.runtime.api.OutputContext;
+import org.apache.tez.runtime.api.events.CompositeDataMovementEvent;
+import org.apache.tez.runtime.api.events.VertexManagerEvent;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.runtime.library.partitioner.HashPartitioner;
+import org.apache.tez.runtime.library.shuffle.impl.ShuffleUserPayloads;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+// Tests which don't require parameterization
+public class TestOrderedPartitionedKVOutput {
+ private Configuration conf;
+ private FileSystem localFs;
+ private Path workingDir;
+
+ @Before
+ public void setup() throws IOException {
+ conf = new Configuration();
+ localFs = FileSystem.getLocal(conf);
+ workingDir =
+ new Path(
+ System.getProperty("test.build.data", System.getProperty("java.io.tmpdir", "/tmp")),
+ TestOrderedPartitionedKVOutput.class.getName())
+ .makeQualified(localFs.getUri(), localFs.getWorkingDirectory());
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, Text.class.getName());
+ conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, Text.class.getName());
+ conf.set(
+ TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, HashPartitioner.class.getName());
+ conf.setStrings(TezRuntimeFrameworkConfigs.LOCAL_DIRS, workingDir.toString());
+ }
+
+ @After
+ public void cleanup() throws IOException {
+ localFs.delete(workingDir, true);
+ }
+
+ @Test(timeout = 5000)
+ public void testNonStartedOutput() throws IOException {
+ OutputContext outputContext = OutputTestHelpers.createOutputContext(conf, conf, workingDir);
+ int numPartitions = 10;
+ OrderedPartitionedKVOutput output =
+ new CelebornOrderedPartitionedKVOutput(outputContext, numPartitions);
+ List<Event> events = output.close();
+ assertEquals(2, events.size());
+ Event event1 = events.get(0);
+ assertTrue(event1 instanceof VertexManagerEvent);
+ Event event2 = events.get(1);
+ assertTrue(event2 instanceof CompositeDataMovementEvent);
+ CompositeDataMovementEvent cdme = (CompositeDataMovementEvent) event2;
+ ByteBuffer bb = cdme.getUserPayload();
+ ShuffleUserPayloads.DataMovementEventPayloadProto shufflePayload =
+ ShuffleUserPayloads.DataMovementEventPayloadProto.parseFrom(ByteString.copyFrom(bb));
+ assertTrue(shufflePayload.hasEmptyPartitions());
+
+ byte[] emptyPartitions =
+ TezCommonUtils.decompressByteStringToByteArray(shufflePayload.getEmptyPartitions());
+ BitSet emptyPartionsBitSet = TezUtilsInternal.fromByteArray(emptyPartitions);
+ assertEquals(numPartitions, emptyPartionsBitSet.cardinality());
+ for (int i = 0; i < numPartitions; i++) {
+ assertTrue(emptyPartionsBitSet.get(i));
+ }
+ }
+
+ // @Test(timeout = 5000)
+ private void testConfigMerge() throws IOException {
+ Configuration localConf = new Configuration(conf);
+ localConf.set("config-from-local", "config-from-local-value");
+ Configuration payload = new Configuration(false);
+ payload.set("config-from-payload", "config-from-payload-value");
+ OutputContext outputContext =
+ OutputTestHelpers.createOutputContext(localConf, payload, workingDir);
+ int numPartitions = 10;
+ OrderedPartitionedKVOutput output =
+ new CelebornOrderedPartitionedKVOutput(outputContext, numPartitions);
+ output.initialize();
+ Configuration configAfterMerge = output.conf;
+ assertEquals("config-from-local-value", configAfterMerge.get("config-from-local"));
+ assertEquals("config-from-payload-value", configAfterMerge.get("config-from-payload"));
+ }
+
+ // @Test(timeout = 10000)
+ private void testClose() throws Exception {
+ OutputContext outputContext = OutputTestHelpers.createOutputContext(conf, conf, workingDir);
+ int numPartitions = 10;
+ OrderedPartitionedKVOutput output =
+ new CelebornOrderedPartitionedKVOutput(outputContext, numPartitions);
+ output.start();
+ output.close();
+ }
+}
diff --git a/pom.xml b/pom.xml
index 59ce95c4b..0f1c30d79 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1154,23 +1154,6 @@
<target>${java.version}</target>
<encoding>UTF-8</encoding>
<fork>true</fork>
- <compilerArgs>
- <arg>-XDcompilePolicy=simple</arg>
- <arg>-Xplugin:ErrorProne \
- -XepExcludedPaths:.*/target/generated-sources/.* \
- -Xep:BadImport:OFF \
- -Xep:EmptyBlockTag:OFF \
- -Xep:EmptyCatch:OFF \
- -Xep:EqualsGetClass:OFF \
- -Xep:FutureReturnValueIgnored:OFF \
- -Xep:JdkObsolete:OFF \
- -Xep:MissingSummary:OFF \
- -Xep:MutableConstantField:OFF \
- -Xep:StringSplitter:OFF \
- -Xep:TypeParameterUnusedInFormals:OFF \
- -Xep:UnnecessaryParentheses:OFF \
- -Xep:UnusedVariable:OFF</arg>
- </compilerArgs>
<annotationProcessorPaths>
<path>
<groupId>com.google.errorprone</groupId>
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezCartesianProductTest.java b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezCartesianProductTest.java
deleted file mode 100644
index d3855cd62..000000000
--- a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezCartesianProductTest.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.celeborn.tests.tez;
-
-import org.apache.commons.lang3.RandomStringUtils;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.util.Tool;
-import org.apache.tez.examples.CartesianProduct;
-import org.junit.jupiter.api.Test;
-
-public class TezCartesianProductTest extends TezIntegrationTestBase {
-
- private String inputPath1 = "cartesian_product_input1";
- private String inputPath2 = "cartesian_product_input2";
- private String inputPath3 = "cartesian_product_input3";
- private String outputPath = "cartesian_product_output";
-
- @Test
- public void cartesianProductTest() throws Exception {
- MiniCelebornCluster.setupMiniClusterWithRandomPorts();
- generateInputFile();
- run();
- }
-
- private void generateInputFile() throws Exception {
- FSDataOutputStream outputStream1 = fs.create(new Path(inputPath1));
- FSDataOutputStream outputStream2 = fs.create(new Path(inputPath2));
- FSDataOutputStream outputStream3 = fs.create(new Path(inputPath3));
- for (int i = 0; i < 500; i++) {
- String alphanumeric = RandomStringUtils.randomAlphanumeric(5);
- String numeric = RandomStringUtils.randomNumeric(5);
- outputStream1.writeBytes(alphanumeric + "\n");
- outputStream2.writeBytes(numeric + "\n");
- if (i % 2 == 0) {
- outputStream3.writeBytes(alphanumeric + "\n");
- }
- }
- outputStream1.close();
- outputStream2.close();
- outputStream3.close();
- }
-
- @Override
- public Tool getTestTool() {
- return new CartesianProduct();
- }
-
- @Override
- public String[] getTestArgs(String uniqueOutputName) {
- return new String[] {
- "-partitioned", inputPath1, inputPath2, inputPath3, outputPath + "/" + uniqueOutputName
- };
- }
-
- @Override
- public String getOutputDir(String uniqueOutputName) {
- return outputPath + "/" + uniqueOutputName;
- }
-
- @Override
- public void verifyResults(String originPath, String rssPath) throws Exception {
- verifyResultsSameSet(originPath, rssPath);
- }
-}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezCartesianProductTest.scala b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezCartesianProductTest.scala
new file mode 100644
index 000000000..bddf7b693
--- /dev/null
+++ b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezCartesianProductTest.scala
@@ -0,0 +1,66 @@
+/*
+ * 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 regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.tests.tez
+
+import org.apache.commons.lang3.RandomStringUtils
+import org.apache.hadoop.fs.FSDataOutputStream
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.util.Tool
+import org.apache.tez.examples.CartesianProduct
+import org.junit.jupiter.api.Test
+
+class TezCartesianProductTest extends TezIntegrationTestBase {
+ private val inputPath1 = "cartesian_product_input1"
+ private val inputPath2 = "cartesian_product_input2"
+ private val inputPath3 = "cartesian_product_input3"
+ private val outputPath = "cartesian_product_output"
+
+ test("celeborn tez integration test - Cartesian Product") {
+ generateInputFile()
+ run()
+ }
+
+ @throws[Exception]
+ private def generateInputFile(): Unit = {
+ val outputStream1 = fs.create(new Path(inputPath1))
+ val outputStream2 = fs.create(new Path(inputPath2))
+ val outputStream3 = fs.create(new Path(inputPath3))
+ for (i <- 0 until 500) {
+ val alphanumeric = RandomStringUtils.randomAlphanumeric(5)
+ val numeric = RandomStringUtils.randomNumeric(5)
+ outputStream1.writeBytes(alphanumeric + "\n")
+ outputStream2.writeBytes(numeric + "\n")
+ if (i % 2 == 0) outputStream3.writeBytes(alphanumeric + "\n")
+ }
+ outputStream1.close()
+ outputStream2.close()
+ outputStream3.close()
+ }
+ override def getTestTool = new CartesianProduct
+ override def getTestArgs(uniqueOutputName: String): Array[String] = Array[String](
+ "-partitioned",
+ inputPath1,
+ inputPath2,
+ inputPath3,
+ outputPath + "/" + uniqueOutputName)
+ override def getOutputDir(uniqueOutputName: String): String = outputPath + "/" + uniqueOutputName
+ @throws[Exception]
+ override def verifyResults(originPath: String, rssPath: String): Unit = {
+ verifyResultsSameSet(originPath, rssPath)
+ }
+}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinBroadcastTest.java b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinBroadcastTest.java
deleted file mode 100644
index 1fe13e3e4..000000000
--- a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinBroadcastTest.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.celeborn.tests.tez;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.util.Tool;
-import org.apache.tez.examples.HashJoinExample;
-import org.junit.jupiter.api.Test;
-
-public class TezHashJoinBroadcastTest extends TezJoinIntegrationTestBase {
-
- private static final String HASH_JOIN_OUTPUT_PATH = "hash_join_output";
-
- @Test
- public void hashJoinDoBroadcastTest() throws Exception {
- MiniCelebornCluster.setupMiniClusterWithRandomPorts();
- generateInputFile();
- String path = HASH_JOIN_OUTPUT_PATH + "_broadcast";
- fs.delete(new Path(path), true);
- run(path);
- }
-
- @Override
- public Tool getTestTool() {
- return new HashJoinExample();
- }
-
- @Override
- public String[] getTestArgs(String uniqueOutputName) {
- if (uniqueOutputName.contains("broadcast")) {
- return new String[] {
- STREAM_INPUT_PATH, HASH_INPUT_PATH, "2", uniqueOutputName, "doBroadcast"
- };
- } else {
- return new String[] {STREAM_INPUT_PATH, HASH_INPUT_PATH, "2", uniqueOutputName};
- }
- }
-
- @Override
- public String getOutputDir(String uniqueOutputName) {
- return uniqueOutputName;
- }
-}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinBroadcastTest.scala b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinBroadcastTest.scala
new file mode 100644
index 000000000..0a408857e
--- /dev/null
+++ b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinBroadcastTest.scala
@@ -0,0 +1,38 @@
+/*
+ * 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 regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.tests.tez
+
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.util.Tool
+import org.apache.tez.examples.HashJoinExample
+import org.junit.jupiter.api.Test
+
+object TezHashJoinBroadcastTest { private val HASH_JOIN_OUTPUT_PATH = "hash_join_output" }
+class TezHashJoinBroadcastTest extends TezJoinIntegrationTestBase {
+
+ test("celeborn tez integration test - Hash Join Broadcast") {
+ generateInputFile()
+ val path = TezHashJoinBroadcastTest.HASH_JOIN_OUTPUT_PATH + "_broadcast"
+ fs.delete(new Path(path), true)
+ run(path)
+ }
+
+ override def getTestTool = new HashJoinExample
+ override def getTestArgs(uniqueOutputName: String): Array[String] = if (uniqueOutputName.contains("broadcast")) Array[String](TezJoinIntegrationTestBase.STREAM_INPUT_PATH, TezJoinIntegrationTestBase.HASH_INPUT_PATH, "2", uniqueOutputName, "doBroadcast") else Array[String](TezJoinIntegrationTestBase.STREAM_INPUT_PATH, TezJoinIntegrationTestBase.HASH_INPUT_PATH, "2", uniqueOutputName)
+ override def getOutputDir(uniqueOutputName: String): String = uniqueOutputName
+}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinTest.java b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinTest.java
deleted file mode 100644
index 99ee3874b..000000000
--- a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinTest.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.celeborn.tests.tez;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.util.Tool;
-import org.apache.tez.examples.HashJoinExample;
-import org.junit.jupiter.api.Test;
-
-public class TezHashJoinTest extends TezJoinIntegrationTestBase {
-
- private static final String HASH_JOIN_OUTPUT_PATH = "hash_join_output";
-
- @Test
- public void hashJoinTest() throws Exception {
- MiniCelebornCluster.setupMiniClusterWithRandomPorts();
- generateInputFile();
- fs.delete(new Path(HASH_JOIN_OUTPUT_PATH), true);
- run(HASH_JOIN_OUTPUT_PATH);
- }
-
- @Override
- public Tool getTestTool() {
- return new HashJoinExample();
- }
-
- @Override
- public String[] getTestArgs(String uniqueOutputName) {
- if (uniqueOutputName.contains("broadcast")) {
- return new String[] {
- STREAM_INPUT_PATH, HASH_INPUT_PATH, "2", uniqueOutputName, "doBroadcast"
- };
- } else {
- return new String[] {STREAM_INPUT_PATH, HASH_INPUT_PATH, "2", uniqueOutputName};
- }
- }
-
- @Override
- public String getOutputDir(String uniqueOutputName) {
- return uniqueOutputName;
- }
-}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinTest.scala b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinTest.scala
new file mode 100644
index 000000000..93fcaecad
--- /dev/null
+++ b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezHashJoinTest.scala
@@ -0,0 +1,37 @@
+/*
+ * 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 regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.tests.tez
+
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.util.Tool
+import org.apache.tez.examples.HashJoinExample
+import org.junit.jupiter.api.Test
+
+object TezHashJoinTest { private val HASH_JOIN_OUTPUT_PATH = "hash_join_output" }
+class TezHashJoinTest extends TezJoinIntegrationTestBase {
+
+ test("celeborn tez integration test - TezHashJoinTest") {
+ generateInputFile()
+ fs.delete(new Path(TezHashJoinTest.HASH_JOIN_OUTPUT_PATH), true)
+ run(TezHashJoinTest.HASH_JOIN_OUTPUT_PATH)
+ }
+
+ override def getTestTool = new HashJoinExample
+ override def getTestArgs(uniqueOutputName: String): Array[String] = if (uniqueOutputName.contains("broadcast")) Array[String](TezJoinIntegrationTestBase.STREAM_INPUT_PATH, TezJoinIntegrationTestBase.HASH_INPUT_PATH, "2", uniqueOutputName, "doBroadcast") else Array[String](TezJoinIntegrationTestBase.STREAM_INPUT_PATH, TezJoinIntegrationTestBase.HASH_INPUT_PATH, "2", uniqueOutputName)
+ override def getOutputDir(uniqueOutputName: String): String = uniqueOutputName
+}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezIntegrationTestBase.scala b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezIntegrationTestBase.scala
new file mode 100644
index 000000000..5394ec8c4
--- /dev/null
+++ b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezIntegrationTestBase.scala
@@ -0,0 +1,297 @@
+/*
+ * 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 regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.tests.tez
+
+import java.io.{EOFException, File}
+import java.util
+import java.util.{ArrayList, HashMap, List, Map, Random}
+
+import scala.collection.JavaConverters._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileStatus, FileSystem, FSDataInputStream, Path}
+import org.apache.hadoop.fs.permission.FsPermission
+import org.apache.hadoop.hdfs.MiniDFSCluster
+import org.apache.hadoop.io.{IntWritable, Text}
+import org.apache.hadoop.util.{LineReader, Tool, ToolRunner}
+import org.apache.tez.client.TezClientUtils
+import org.apache.tez.dag.api.TezConfiguration
+import org.apache.tez.dag.api.TezConfiguration.TEZ_LIB_URIS
+import org.apache.tez.dag.app.CelebornDagAppMaster
+import org.apache.tez.test.MiniTezCluster
+import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
+import org.scalatest.funsuite.AnyFunSuite
+
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.util.Utils
+import org.apache.celeborn.service.deploy.MiniClusterFeature
+import org.apache.celeborn.service.deploy.master.Master
+import org.apache.celeborn.service.deploy.worker.Worker
+import org.apache.celeborn.tez.plugin.util.CelebornTezUtils
+
+class TezIntegrationTestBase extends AnyFunSuite with Logging with MiniClusterFeature
+ with BeforeAndAfterAll with BeforeAndAfterEach {
+
+ private val TEST_ROOT_DIR =
+ "target" + Path.SEPARATOR + classOf[TezIntegrationTestBase].getName + "-tmpDir"
+
+ protected var fs: FileSystem = null
+ var conf: Configuration = null
+ protected var remoteStagingDir: Path = null
+ protected var cluster: MiniDFSCluster = null
+ protected var HDFS_URI: String = null
+ var miniTezCluster: MiniTezCluster = null
+ var workers: collection.Set[Worker] = null
+ var master: Master = null
+
+ override def beforeAll(): Unit = {
+ logInfo("test initialized , setup celeborn mini cluster")
+ val (newMaster, newWorkers) = setupMiniClusterWithRandomPorts()
+ master = newMaster
+ workers = newWorkers
+
+ // hdfs
+ this.conf = new Configuration
+ val output = Utils.createTempDir(System.getProperty("java.io.tmpdir"), "output")
+ logInfo(s"Output tmp dir: $output")
+ conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, output.getAbsolutePath)
+ cluster = new MiniDFSCluster.Builder(conf).build
+ HDFS_URI = cluster.getURI.toString + "/"
+ fs = new Path(HDFS_URI).getFileSystem(conf)
+
+ // tez
+ miniTezCluster = new MiniTezCluster(classOf[TezIntegrationTestBase].getName, 1, 1, 1)
+ miniTezCluster.init(conf)
+ miniTezCluster.start()
+ }
+
+ override def afterAll(): Unit = {
+ logInfo("all test complete , stop celeborn mini cluster")
+ shutdownMiniCluster()
+ if (miniTezCluster != null) {
+ miniTezCluster.stop()
+ }
+ fs.close()
+ cluster.shutdown()
+ }
+
+ override def beforeEach(): Unit = {
+ remoteStagingDir =
+ fs.makeQualified(new Path(TEST_ROOT_DIR, String.valueOf(new Random().nextInt(100000))))
+ TezClientUtils.ensureStagingDirExists(conf, remoteStagingDir)
+ }
+
+ @throws[Exception]
+ def run(): Unit = {
+ val appConf = new TezConfiguration(miniTezCluster.getConfig());
+ updateCommonConfiguration(appConf);
+ runTezApp(appConf, getTestTool, getTestArgs("origin"));
+ val originPath = getOutputDir("origin")
+ runRssTest("celeborn", originPath)
+ }
+
+ @throws[Exception]
+ private def runRssTest(testName: String, originPath: String): Unit = {
+ val appConf = new TezConfiguration(miniTezCluster.getConfig)
+ updateRssConfiguration(appConf)
+ appendAndUploadRssJars(appConf)
+ runTezApp(appConf, getTestTool, getTestArgs(testName))
+ verifyResults(originPath, getOutputDir(testName))
+ }
+
+ @throws[Exception]
+ protected def runTezApp(tezConf: TezConfiguration, tool: Tool, args: Array[String]): Unit = {
+ assertEquals(0, ToolRunner.run(tezConf, tool, args), tool.getClass.getName + " failed")
+ }
+
+ protected def appendAndUploadRssJars(tezConf: TezConfiguration): Unit = {
+ val uris = tezConf.get(TEZ_LIB_URIS)
+ Assertions.assertNotNull(uris)
+ // Get the rss client tez shaded jar file.
+ val url = classOf[TezIntegrationTestBase].getResource("/")
+ val parentPath = new Path(url.getPath).getParent.getParent.getParent.getParent.toString
+ val file = new File(parentPath, "client-tez/tez-shaded/target")
+ val jars = file.listFiles
+ var rssJar: File = null
+ for (jar <- jars) {
+ if (rssJar == null && jar.getName.startsWith("celeborn-client-tez-shaded")) {
+ rssJar = jar
+ }
+ }
+ // upload rss jars
+ val testRootDir =
+ fs.makeQualified(new Path("target", classOf[TezIntegrationTestBase].getName + "-tmpDir"))
+ val appRemoteJar = new Path(testRootDir, "rss-tez-client-shaded.jar")
+ fs.copyFromLocalFile(new Path(rssJar.toString), appRemoteJar)
+ fs.setPermission(appRemoteJar, new FsPermission("777"))
+ // update tez.lib.uris
+ tezConf.set(TEZ_LIB_URIS, uris + "," + appRemoteJar)
+ }
+
+ def getOutputDir(uniqueOutputName: String): String = {
+ Assertions.fail("getOutputDir is not implemented")
+ null
+ }
+
+ def getTestTool: Tool = null
+
+ def getTestArgs(uniqueOutputName: String) = new Array[String](0)
+
+ def updateCommonConfiguration(appConf: Configuration): Unit = {
+ appConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString)
+ appConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 512)
+ appConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx384m")
+ appConf.setInt(TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB, 512)
+ appConf.set(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS, " -Xmx384m")
+ }
+
+ @throws[Exception]
+ def updateRssConfiguration(appConf: Configuration): Unit = {
+ appConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString);
+ appConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 512)
+ appConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx384m")
+ appConf.setInt(TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB, 512)
+ appConf.set(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS, " -Xmx384m")
+ appConf.set(
+ CelebornTezUtils.TEZ_PREFIX + CelebornConf.MASTER_ENDPOINTS.key,
+ s"localhost:${master.conf.get(CelebornConf.MASTER_PORT)}")
+ appConf.set(
+ TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS,
+ TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS_DEFAULT + " " + classOf[CelebornDagAppMaster].getName)
+ }
+
+ @throws[Exception]
+ def verifyResults(originPath: String, rssPath: String): Unit = {
+ verifyResultEqual(originPath, rssPath)
+ }
+
+ @throws[Exception]
+ def verifyResultEqual(originPath: String, rssPath: String): Unit = {
+ if (originPath == null && rssPath == null) return
+ val originPathFs: Path = new Path(originPath)
+ val rssPathFs: Path = new Path(rssPath)
+ val originFiles: Array[FileStatus] = fs.listStatus(originPathFs)
+ val rssFiles: Array[FileStatus] = fs.listStatus(rssPathFs)
+ var originLen: Long = 0
+ var rssLen: Long = 0
+ val originFileList: util.List[String] = new util.ArrayList[String]
+ val rssFileList: util.List[String] = new util.ArrayList[String]
+ for (file <- originFiles) {
+ originLen += file.getLen
+ val name: String = file.getPath.getName
+ if (!(name == "_SUCCESS")) originFileList.add(name)
+ }
+ for (file <- rssFiles) {
+ rssLen += file.getLen
+ val name: String = file.getPath.getName
+ if (!(name == "_SUCCESS")) rssFileList.add(name)
+ }
+ assertEquals(originFileList.size, rssFileList.size)
+ for (i <- 0 until originFileList.size) {
+ assertEquals(originFileList.get(i), rssFileList.get(i))
+ val p1: Path = new Path(originPath, originFileList.get(i))
+ val f1: FSDataInputStream = fs.open(p1)
+ val p2: Path = new Path(rssPath, rssFileList.get(i))
+ val f2: FSDataInputStream = fs.open(p2)
+ var isNotEof1: Boolean = true
+ var isNotEof2: Boolean = true
+ while (isNotEof1 && isNotEof2) {
+ var b1: Byte = 1
+ var b2: Byte = 1
+ try b1 = f1.readByte
+ catch {
+ case ee: EOFException =>
+ isNotEof1 = false
+ }
+ try b2 = f2.readByte
+ catch {
+ case ee: EOFException =>
+ isNotEof2 = false
+ }
+ assertEquals(b1, b2)
+ }
+ assertEquals(isNotEof1, isNotEof2)
+ }
+ assertEquals(originLen, rssLen)
+ }
+
+ @throws[Exception]
+ def verifyResultsSameSet(originPath: String, rssPath: String): Unit = {
+ if (originPath == null && rssPath == null) return
+ // 1 List the originalPath and rssPath, make sure generated file are same.
+ val originPathFs: Path = new Path(originPath)
+ val rssPathFs: Path = new Path(rssPath)
+ val originFiles: Array[FileStatus] = fs.listStatus(originPathFs)
+ val rssFiles: Array[FileStatus] = fs.listStatus(rssPathFs)
+ var originLen: Long = 0
+ var rssLen: Long = 0
+ val originFileList: util.List[String] = new util.ArrayList[String]
+ val rssFileList: util.List[String] = new util.ArrayList[String]
+ for (file <- originFiles) {
+ originLen += file.getLen
+ val name: String = file.getPath.getName
+ if (!(name == "_SUCCESS")) originFileList.add(name)
+ }
+ for (file <- rssFiles) {
+ rssLen += file.getLen
+ val name: String = file.getPath.getName
+ if (!(name == "_SUCCESS")) rssFileList.add(name)
+ }
+ assertEquals(originFileList.size, rssFileList.size)
+ // 2 Load original result and rss result to hashmap
+ val originalResults: util.Map[String, Integer] = new util.HashMap[String, Integer]
+ import scala.collection.JavaConversions._
+ for (file <- originFileList) {
+ val path: Path = new Path(originPath, file)
+ val lineReader: LineReader = new LineReader(fs.open(path))
+ val line: Text = new Text
+ while (lineReader.readLine(line) > 0)
+ if (!originalResults.containsKey(line.toString)) originalResults.put(line.toString, 1)
+ else originalResults.put(line.toString, originalResults.get(line.toString) + 1)
+ }
+ val rssResults: util.Map[String, Integer] = new util.HashMap[String, Integer]
+ import scala.collection.JavaConversions._
+ for (file <- rssFileList) {
+ val path: Path = new Path(rssPath, file)
+ val lineReader: LineReader = new LineReader(fs.open(path))
+ val line: Text = new Text
+ while (lineReader.readLine(line) > 0)
+ if (!rssResults.containsKey(line.toString)) rssResults.put(line.toString, 1)
+ else rssResults.put(line.toString, rssResults.get(line.toString) + 1)
+ }
+ // 3 Compare the hashmap
+ Assertions.assertEquals(
+ originalResults.size,
+ rssResults.size,
+ "The size of cartesian product set is not equal")
+ import scala.collection.JavaConversions._
+ for (entry <- originalResults.entrySet) {
+ Assertions.assertTrue(
+ rssResults.containsKey(entry.getKey),
+ entry.getKey + " is not found in rss cartesian product result")
+ Assertions.assertEquals(
+ entry.getValue,
+ rssResults.get(entry.getKey),
+ "the value of " + entry.getKey + " is not equal to in rss cartesian product result")
+ }
+ }
+
+}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezIntegrationTestBase.java b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezIntegrationTestBase2.java
similarity index 94%
rename from tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezIntegrationTestBase.java
rename to tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezIntegrationTestBase2.java
index 65a1cfeaf..08113bec9 100644
--- a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezIntegrationTestBase.java
+++ b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezIntegrationTestBase2.java
@@ -37,7 +37,6 @@ import org.apache.hadoop.util.ToolRunner;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.app.CelebornDagAppMaster;
import org.apache.tez.test.MiniTezCluster;
-import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.slf4j.Logger;
@@ -46,7 +45,7 @@ import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.CelebornConf;
import org.apache.celeborn.tez.plugin.util.CelebornTezUtils;
-public class TezIntegrationTestBase {
+public class TezIntegrationTestBase2 {
private static final Logger LOG = LoggerFactory.getLogger(TezIntegrationTestBase.class);
protected static MiniTezCluster miniTezCluster;
@@ -62,92 +61,32 @@ public class TezIntegrationTestBase {
fs = FileSystem.get(conf);
}
- @AfterAll
- public static void afterClass() throws Exception {
- MiniCelebornCluster.shutdownMiniCluster();
- }
-
- public static void verifyResultsSameSet(String originPath, String rssPath) throws Exception {
- if (originPath == null && rssPath == null) {
- return;
- }
- // 1 List the originalPath and rssPath, make sure generated file are same.
- Path originPathFs = new Path(originPath);
- Path rssPathFs = new Path(rssPath);
- FileStatus[] originFiles = fs.listStatus(originPathFs);
- FileStatus[] rssFiles = fs.listStatus(rssPathFs);
- long originLen = 0;
- long rssLen = 0;
- List<String> originFileList = new ArrayList<>();
- List<String> rssFileList = new ArrayList<>();
- for (FileStatus file : originFiles) {
- originLen += file.getLen();
- String name = file.getPath().getName();
- if (!name.equals("_SUCCESS")) {
- originFileList.add(name);
- }
- }
- for (FileStatus file : rssFiles) {
- rssLen += file.getLen();
- String name = file.getPath().getName();
- if (!name.equals("_SUCCESS")) {
- rssFileList.add(name);
- }
- }
- assertEquals(originFileList.size(), rssFileList.size());
+ public void run() throws Exception {
- // 2 Load original result and rss result to hashmap
- Map<String, Integer> originalResults = new HashMap<>();
- for (String file : originFileList) {
- Path path = new Path(originPath, file);
- LineReader lineReader = new LineReader(fs.open(path));
- Text line = new Text();
- while (lineReader.readLine(line) > 0) {
- if (!originalResults.containsKey(line.toString())) {
- originalResults.put(line.toString(), 1);
- } else {
- originalResults.put(line.toString(), originalResults.get(line.toString()) + 1);
- }
- }
- }
+ // TezConfiguration appConf = new TezConfiguration(miniTezCluster.getConfig());
+ TezConfiguration appConf = new TezConfiguration();
+ updateCommonConfiguration(appConf);
+ // runTezApp(appConf, getTestTool(), getTestArgs("test"));
- Map<String, Integer> rssResults = new HashMap<>();
- for (String file : rssFileList) {
- Path path = new Path(rssPath, file);
- LineReader lineReader = new LineReader(fs.open(path));
- Text line = new Text();
- while (lineReader.readLine(line) > 0) {
- if (!rssResults.containsKey(line.toString())) {
- rssResults.put(line.toString(), 1);
- } else {
- rssResults.put(line.toString(), rssResults.get(line.toString()) + 1);
- }
- }
- }
+ appConf = new TezConfiguration();
+ updateRssConfiguration(appConf);
+ runTezApp(appConf, getTestTool(), getTestArgs("rss"));
+ }
- // 3 Compare the hashmap
- Assertions.assertEquals(
- originalResults.size(),
- rssResults.size(),
- "The size of cartesian product set is not equal");
- for (Map.Entry<String, Integer> entry : originalResults.entrySet()) {
- Assertions.assertTrue(
- rssResults.containsKey(entry.getKey()),
- entry.getKey() + " is not found in rss cartesian product result");
- Assertions.assertEquals(
- entry.getValue(),
- rssResults.get(entry.getKey()),
- "the value of " + entry.getKey() + " is not equal to in rss cartesian product result");
- }
+ protected void runTezApp(TezConfiguration tezConf, Tool tool, String[] args) throws Exception {
+ ToolRunner.run(tezConf, tool, args);
}
- public void run() throws Exception {
- TezConfiguration appConf = new TezConfiguration();
- updateRssConfiguration(appConf);
- runTezApp(appConf, getTestTool(), getTestArgs("rss"));
+ public void updateCommonConfiguration(Configuration appConf) throws Exception {
+ // appConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
+ appConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 512);
+ appConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx384m");
+ appConf.setInt(TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB, 512);
+ appConf.set(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS, " -Xmx384m");
}
public void updateRssConfiguration(Configuration appConf) throws Exception {
+ // appConf.set(TezConfiguration.TEZ_AM_STAGING_DIR, remoteStagingDir.toString());
appConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 512);
appConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx384m");
appConf.set(
@@ -155,10 +94,8 @@ public class TezIntegrationTestBase {
"org.apache.tez.dag.app.CelebornDagAppMaster");
appConf.setInt(TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB, 512);
appConf.set(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS, " -Xmx384m");
-
appConf.set(
- CelebornTezUtils.TEZ_PREFIX + CelebornConf.MASTER_ENDPOINTS().key(),
- "localhost:" + MiniCelebornCluster.master.conf().get(CelebornConf.MASTER_PORT()));
+ CelebornTezUtils.TEZ_PREFIX + CelebornConf.MASTER_ENDPOINTS().key(), "localhost:9097");
appConf.set(
TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS,
TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS_DEFAULT
@@ -166,10 +103,6 @@ public class TezIntegrationTestBase {
+ CelebornDagAppMaster.class.getName());
}
- protected void runTezApp(TezConfiguration tezConf, Tool tool, String[] args) throws Exception {
- ToolRunner.run(tezConf, tool, args);
- }
-
public Tool getTestTool() {
return null;
}
@@ -178,13 +111,6 @@ public class TezIntegrationTestBase {
return new String[0];
}
- public void updateCommonConfiguration(Configuration appConf) throws Exception {
- appConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 512);
- appConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx384m");
- appConf.setInt(TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB, 512);
- appConf.set(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS, " -Xmx384m");
- }
-
public String getOutputDir(String uniqueOutputName) {
return null;
}
@@ -247,4 +173,78 @@ public class TezIntegrationTestBase {
}
assertEquals(originLen, rssLen);
}
+
+ public static void verifyResultsSameSet(String originPath, String rssPath) throws Exception {
+ if (originPath == null && rssPath == null) {
+ return;
+ }
+ // 1 List the originalPath and rssPath, make sure generated file are same.
+ Path originPathFs = new Path(originPath);
+ Path rssPathFs = new Path(rssPath);
+ FileStatus[] originFiles = fs.listStatus(originPathFs);
+ FileStatus[] rssFiles = fs.listStatus(rssPathFs);
+ long originLen = 0;
+ long rssLen = 0;
+ List<String> originFileList = new ArrayList<>();
+ List<String> rssFileList = new ArrayList<>();
+ for (FileStatus file : originFiles) {
+ originLen += file.getLen();
+ String name = file.getPath().getName();
+ if (!name.equals("_SUCCESS")) {
+ originFileList.add(name);
+ }
+ }
+ for (FileStatus file : rssFiles) {
+ rssLen += file.getLen();
+ String name = file.getPath().getName();
+ if (!name.equals("_SUCCESS")) {
+ rssFileList.add(name);
+ }
+ }
+ assertEquals(originFileList.size(), rssFileList.size());
+
+ // 2 Load original result and rss result to hashmap
+ Map<String, Integer> originalResults = new HashMap<>();
+ for (String file : originFileList) {
+ Path path = new Path(originPath, file);
+ LineReader lineReader = new LineReader(fs.open(path));
+ Text line = new Text();
+ while (lineReader.readLine(line) > 0) {
+ if (!originalResults.containsKey(line.toString())) {
+ originalResults.put(line.toString(), 1);
+ } else {
+ originalResults.put(line.toString(), originalResults.get(line.toString()) + 1);
+ }
+ }
+ }
+
+ Map<String, Integer> rssResults = new HashMap<>();
+ for (String file : rssFileList) {
+ Path path = new Path(rssPath, file);
+ LineReader lineReader = new LineReader(fs.open(path));
+ Text line = new Text();
+ while (lineReader.readLine(line) > 0) {
+ if (!rssResults.containsKey(line.toString())) {
+ rssResults.put(line.toString(), 1);
+ } else {
+ rssResults.put(line.toString(), rssResults.get(line.toString()) + 1);
+ }
+ }
+ }
+
+ // 3 Compare the hashmap
+ Assertions.assertEquals(
+ originalResults.size(),
+ rssResults.size(),
+ "The size of cartesian product set is not equal");
+ for (Map.Entry<String, Integer> entry : originalResults.entrySet()) {
+ Assertions.assertTrue(
+ rssResults.containsKey(entry.getKey()),
+ entry.getKey() + " is not found in rss cartesian product result");
+ Assertions.assertEquals(
+ entry.getValue(),
+ rssResults.get(entry.getKey()),
+ "the value of " + entry.getKey() + " is not equal to in rss cartesian product result");
+ }
+ }
}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezJoinIntegrationTestBase.java b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezJoinIntegrationTestBase.java
deleted file mode 100644
index 291369f8e..000000000
--- a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezJoinIntegrationTestBase.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.celeborn.tests.tez;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.util.ToolRunner;
-import org.apache.tez.dag.api.TezConfiguration;
-import org.apache.tez.examples.JoinDataGen;
-import org.apache.tez.examples.JoinValidate;
-
-public class TezJoinIntegrationTestBase extends TezIntegrationTestBase {
-
- protected static final String STREAM_INPUT_PATH = "stream_input";
- protected static final String STREAM_INPUT_FILE_SIZE = "5000000";
- protected static final String HASH_INPUT_PATH = "hash_input";
- protected static final String HASH_INPUT_FILE_SIZE = "500000";
- protected static final String JOIN_EXPECTED_PATH = "join_expected";
- protected static final String NUM_TASKS = "2";
-
- protected void generateInputFile() throws Exception {
-
- fs.delete(new Path(STREAM_INPUT_PATH), true);
- fs.delete(new Path(HASH_INPUT_PATH), true);
- fs.delete(new Path(JOIN_EXPECTED_PATH), true);
- String[] args = {
- STREAM_INPUT_PATH,
- STREAM_INPUT_FILE_SIZE,
- HASH_INPUT_PATH,
- HASH_INPUT_FILE_SIZE,
- JOIN_EXPECTED_PATH,
- NUM_TASKS
- };
- JoinDataGen dataGen = new JoinDataGen();
- TezConfiguration appConf = new TezConfiguration(miniTezCluster.getConfig());
- updateCommonConfiguration(appConf);
- assertEquals(0, ToolRunner.run(appConf, dataGen, args), "JoinDataGen failed");
- }
-
- public void run(String path) throws Exception {
- TezConfiguration appConf = new TezConfiguration();
- updateRssConfiguration(appConf);
- runTezApp(appConf, getTestTool(), getTestArgs(path));
- verifyResults(JOIN_EXPECTED_PATH, getOutputDir(path));
- }
-
- public void verifyResults(String expectedPath, String rssPath) throws Exception {
- String[] args = {expectedPath, rssPath};
- JoinValidate validate = new JoinValidate();
- TezConfiguration appConf = new TezConfiguration(miniTezCluster.getConfig());
- updateCommonConfiguration(appConf);
- assertEquals(0, ToolRunner.run(appConf, validate, args), "JoinValidate failed");
- }
-}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezJoinIntegrationTestBase.scala b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezJoinIntegrationTestBase.scala
new file mode 100644
index 000000000..5445a6d55
--- /dev/null
+++ b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezJoinIntegrationTestBase.scala
@@ -0,0 +1,72 @@
+/*
+ * 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 regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.tests.tez
+
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.util.ToolRunner
+import org.apache.tez.dag.api.TezConfiguration
+import org.apache.tez.examples.JoinDataGen
+import org.apache.tez.examples.JoinValidate
+import org.junit.jupiter.api.Assertions.assertEquals
+
+object TezJoinIntegrationTestBase {
+ val STREAM_INPUT_PATH = "stream_input"
+ val STREAM_INPUT_FILE_SIZE = "5000000"
+ val HASH_INPUT_PATH = "hash_input"
+ val HASH_INPUT_FILE_SIZE = "500000"
+ val JOIN_EXPECTED_PATH = "join_expected"
+ val NUM_TASKS = "2"
+}
+class TezJoinIntegrationTestBase extends TezIntegrationTestBase {
+ @throws[Exception]
+ protected def generateInputFile(): Unit = {
+ fs.delete(new Path(TezJoinIntegrationTestBase.STREAM_INPUT_PATH), true)
+ fs.delete(new Path(TezJoinIntegrationTestBase.HASH_INPUT_PATH), true)
+ fs.delete(new Path(TezJoinIntegrationTestBase.JOIN_EXPECTED_PATH), true)
+ val args = Array(
+ TezJoinIntegrationTestBase.STREAM_INPUT_PATH,
+ TezJoinIntegrationTestBase.STREAM_INPUT_FILE_SIZE,
+ TezJoinIntegrationTestBase.HASH_INPUT_PATH,
+ TezJoinIntegrationTestBase.HASH_INPUT_FILE_SIZE,
+ TezJoinIntegrationTestBase.JOIN_EXPECTED_PATH,
+ TezJoinIntegrationTestBase.NUM_TASKS)
+ val dataGen = new JoinDataGen
+ val appConf = new TezConfiguration(miniTezCluster.getConfig)
+ updateCommonConfiguration(appConf)
+ assertEquals(0, ToolRunner.run(appConf, dataGen, args), "JoinDataGen failed")
+ }
+
+ @throws[Exception]
+ def run(path: String): Unit = {
+ val appConf = new TezConfiguration(miniTezCluster.getConfig());
+ updateRssConfiguration(appConf)
+ appendAndUploadRssJars(appConf)
+ runTezApp(appConf, getTestTool, getTestArgs(path))
+ verifyResults(TezJoinIntegrationTestBase.JOIN_EXPECTED_PATH, getOutputDir(path))
+ }
+
+
+ @throws[Exception]
+ override def verifyResults(expectedPath: String, rssPath: String): Unit = {
+ val args = Array(expectedPath, rssPath)
+ val validate = new JoinValidate
+ val appConf = new TezConfiguration(miniTezCluster.getConfig)
+ updateCommonConfiguration(appConf)
+ assertEquals(0, ToolRunner.run(appConf, validate, args), "JoinValidate failed")
+ }
+}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezOrderedWordCountTest.java b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezOrderedWordCountTest.java
deleted file mode 100644
index 21f45169f..000000000
--- a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezOrderedWordCountTest.java
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.celeborn.tests.tez;
-
-import java.util.*;
-
-import com.google.common.collect.Lists;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.util.Tool;
-import org.apache.tez.examples.OrderedWordCount;
-import org.junit.jupiter.api.Test;
-
-public class TezOrderedWordCountTest extends TezIntegrationTestBase {
-
- private String inputPath = "ordered_word_count_input";
- private String outputPath = "ordered_word_count_output";
- private List<String> wordTable =
- Lists.newArrayList(
- "apple", "banana", "fruit", "cherry", "Chinese", "America", "Japan", "tomato");
-
- @Test
- public void orderedWordCountTest() throws Exception {
- MiniCelebornCluster.setupMiniClusterWithRandomPorts();
- generateInputFile();
- run();
- }
-
- private void generateInputFile() throws Exception {
- // For ordered word count, the key of last ordered sorter is the summation of word, the value is
- // the word. So it means this key may not be unique. Because Sorter can only make sure key is
- // sorted, so the second column (word column) may be not sorted.
- // To keep pace with verifyResults, here make sure summation of word is unique number.
- FSDataOutputStream outputStream = fs.create(new Path(inputPath));
- Random random = new Random();
- Set<Integer> used = new HashSet();
- List<String> outputList = new ArrayList<>();
- int index = 0;
- while (index < wordTable.size()) {
- int summation = random.nextInt(50);
- if (used.contains(summation)) {
- continue;
- }
- used.add(summation);
- for (int i = 0; i < summation; i++) {
- outputList.add(wordTable.get(index));
- }
- index++;
- }
- Collections.shuffle(outputList);
- for (String word : outputList) {
- outputStream.writeBytes(word + "\n");
- }
- outputStream.close();
- }
-
- @Override
- public Tool getTestTool() {
- return new OrderedWordCount();
- }
-
- @Override
- public String[] getTestArgs(String uniqueOutputName) {
- return new String[] {inputPath, outputPath + "/" + uniqueOutputName, "2"};
- }
-
- @Override
- public String getOutputDir(String uniqueOutputName) {
- return outputPath + "/" + uniqueOutputName;
- }
-}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezOrderedWordCountTest.scala b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezOrderedWordCountTest.scala
new file mode 100644
index 000000000..673acf06f
--- /dev/null
+++ b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezOrderedWordCountTest.scala
@@ -0,0 +1,76 @@
+/*
+ * 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 regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.tests.tez
+
+import java.util._
+import com.google.common.collect.Lists
+import org.apache.hadoop.fs.FSDataOutputStream
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.util.Tool
+import org.apache.tez.examples.OrderedWordCount
+import org.junit.jupiter.api.Test
+
+import java.util
+
+class TezOrderedWordCountTest extends TezIntegrationTestBase {
+ private val inputPath = "ordered_word_count_input"
+ private val outputPath = "ordered_word_count_output"
+ private val wordTable = Lists.newArrayList(
+ "apple",
+ "banana",
+ "fruit",
+ "cherry",
+ "Chinese",
+ "America",
+ "Japan",
+ "tomato")
+
+ test("celeborn tez integration test - Ordered Word Count") {
+ generateInputFile()
+ run()
+ }
+
+ @throws[Exception]
+ private def generateInputFile(): Unit = {
+ // For ordered word count, the key of last ordered sorter is the summation of word, the value is
+ // the word. So it means this key may not be unique. Because Sorter can only make sure key is
+ // sorted, so the second column (word column) may be not sorted.
+ // To keep pace with verifyResults, here make sure summation of word is unique number.
+ val outputStream = fs.create(new Path(inputPath))
+ val random = new Random
+ val used = new util.HashSet[Int]()
+ val outputList = new util.ArrayList[String]
+ var index = 0
+ while (index < wordTable.size) {
+ val summation = random.nextInt(50)
+ if (!used.contains(summation)) {
+ used.add(summation)
+ for (i <- 0 until summation) { outputList.add(wordTable.get(index)) }
+ index += 1
+ }
+
+ }
+ Collections.shuffle(outputList)
+ import scala.collection.JavaConversions._
+ for (word <- outputList) { outputStream.writeBytes(word + "\n") }
+ outputStream.close()
+ }
+ override def getTestTool = new OrderedWordCount
+ override def getTestArgs(uniqueOutputName: String): Array[String] = Array[String](inputPath, outputPath + "/" + uniqueOutputName, "2")
+ override def getOutputDir(uniqueOutputName: String): String = outputPath + "/" + uniqueOutputName
+}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezSortMergeJoinTest.java b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezSortMergeJoinTest.java
deleted file mode 100644
index 5bfb3743e..000000000
--- a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezSortMergeJoinTest.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.celeborn.tests.tez;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.util.Tool;
-import org.apache.tez.examples.SortMergeJoinExample;
-import org.junit.jupiter.api.Test;
-
-public class TezSortMergeJoinTest extends TezJoinIntegrationTestBase {
-
- private static final String SORT_MERGE_JOIN_OUTPUT_PATH = "sort_merge_join_output";
-
- @Test
- public void sortMergeJoinTest() throws Exception {
- MiniCelebornCluster.setupMiniClusterWithRandomPorts();
- fs.delete(new Path(SORT_MERGE_JOIN_OUTPUT_PATH), true);
- generateInputFile();
- run(SORT_MERGE_JOIN_OUTPUT_PATH);
- }
-
- @Override
- public Tool getTestTool() {
- return new SortMergeJoinExample();
- }
-
- @Override
- public String[] getTestArgs(String uniqueOutputName) {
- return new String[] {STREAM_INPUT_PATH, HASH_INPUT_PATH, "2", uniqueOutputName};
- }
-
- @Override
- public String getOutputDir(String uniqueOutputName) {
- return uniqueOutputName;
- }
-}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezSortMergeJoinTest.scala b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezSortMergeJoinTest.scala
new file mode 100644
index 000000000..ee67a3244
--- /dev/null
+++ b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezSortMergeJoinTest.scala
@@ -0,0 +1,38 @@
+/*
+ * 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 regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.tests.tez
+
+import org.apache.hadoop.fs.Path
+import org.apache.tez.examples.SortMergeJoinExample
+import org.junit.jupiter.api.Test
+
+object TezSortMergeJoinTest {
+ private val SORT_MERGE_JOIN_OUTPUT_PATH = "sort_merge_join_output"
+}
+class TezSortMergeJoinTest extends TezJoinIntegrationTestBase {
+
+ test("celeborn tez integration test - Sort Merge Join") {
+ fs.delete(new Path(TezSortMergeJoinTest.SORT_MERGE_JOIN_OUTPUT_PATH), true)
+ generateInputFile()
+ run(TezSortMergeJoinTest.SORT_MERGE_JOIN_OUTPUT_PATH)
+ }
+
+ override def getTestTool = new SortMergeJoinExample
+ override def getTestArgs(uniqueOutputName: String): Array[String] = Array[String](TezJoinIntegrationTestBase.STREAM_INPUT_PATH, TezJoinIntegrationTestBase.HASH_INPUT_PATH, "2", uniqueOutputName)
+ override def getOutputDir(uniqueOutputName: String): String = uniqueOutputName
+}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezWordCountTest.java b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezWordCountTest.java
deleted file mode 100644
index b6c940f1d..000000000
--- a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezWordCountTest.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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 regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.celeborn.tests.tez;
-
-import java.io.FileOutputStream;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.Random;
-
-import com.google.common.collect.Lists;
-import org.apache.hadoop.util.Tool;
-import org.apache.tez.examples.WordCount;
-import org.junit.Test;
-
-public class TezWordCountTest extends TezIntegrationTestBase {
-
- private String inputPath = "word_count_input";
- private String outputPath = "word_count_output";
- private List<String> wordTable =
- Lists.newArrayList(
- "apple", "banana", "fruit", "cherry", "Chinese", "America", "Japan", "tomato");
-
- @Test
- public void wordCountTest() throws Exception {
- MiniCelebornCluster.setupMiniClusterWithRandomPorts();
- generateInputFile();
- run();
- }
-
- private void generateInputFile() throws Exception {
- FileOutputStream outputStream = new FileOutputStream(inputPath);
-
- Random random = new Random();
- for (int i = 0; i < 100; i++) {
- int index = random.nextInt(wordTable.size());
- String str = wordTable.get(index) + "\n";
- outputStream.write(str.getBytes(StandardCharsets.UTF_8));
- }
- outputStream.close();
- }
-
- @Override
- public Tool getTestTool() {
- return new WordCount();
- }
-
- @Override
- public String[] getTestArgs(String uniqueOutputName) {
- return new String[] {inputPath, outputPath + "/" + uniqueOutputName, "2"};
- }
-
- @Override
- public String getOutputDir(String uniqueOutputName) {
- return outputPath + "/" + uniqueOutputName;
- }
-}
diff --git a/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezWordCountTest.scala b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezWordCountTest.scala
new file mode 100644
index 000000000..d9925b35e
--- /dev/null
+++ b/tests/tez-it/src/test/java/org/apache/celeborn/tests/tez/TezWordCountTest.scala
@@ -0,0 +1,61 @@
+/*
+ * 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 regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.tests.tez
+
+import java.io.FileOutputStream
+import java.nio.charset.StandardCharsets
+import java.util.Random
+
+import com.google.common.collect.Lists
+import org.apache.hadoop.fs.Path
+import org.apache.tez.examples.WordCount
+
+class TezWordCountTest extends TezIntegrationTestBase {
+ private val inputPath = "word_count_input"
+ private val outputPath = "word_count_output"
+ private val wordTable = Lists.newArrayList(
+ "apple",
+ "banana",
+ "fruit",
+ "cherry",
+ "Chinese",
+ "America",
+ "Japan",
+ "tomato")
+
+ test("celeborn tez integration test - Word Count") {
+ generateInputFile()
+ run()
+ }
+
+ @throws[Exception]
+ private def generateInputFile(): Unit = {
+ val outputStream = fs.create(new Path(inputPath))
+ val random = new Random
+ for (i <- 0 until 100) {
+ val index = random.nextInt(wordTable.size)
+ val str = wordTable.get(index) + "\n"
+ outputStream.write(str.getBytes(StandardCharsets.UTF_8))
+ }
+ outputStream.close()
+ }
+ override def getTestTool = new WordCount
+ override def getTestArgs(uniqueOutputName: String): Array[String] =
+ Array[String](inputPath, outputPath + "/" + uniqueOutputName, "2")
+ override def getOutputDir(uniqueOutputName: String): String = outputPath + "/" + uniqueOutputName
+}
diff --git a/tests/tez-it/src/test/resources/tez-site.xml b/tests/tez-it/src/test/resources/tez-site.xml
index a2ccf2bc4..c7f5b822b 100644
--- a/tests/tez-it/src/test/resources/tez-site.xml
+++ b/tests/tez-it/src/test/resources/tez-site.xml
@@ -18,10 +18,4 @@
<!-- Put site-specific property overrides in this file. -->
<configuration>
-
-<property>
- <name>tez.local.mode</name>
- <value>true</value>
-</property>
-
</configuration>