Skip to content

TEZ-4547: Add Tez AM JobID to the JobConf #339

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
abstractdog merged 8 commits into from
Aug 5, 2024
Merged

TEZ-4547: Add Tez AM JobID to the JobConf #339

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9040a2f
TEZ-4547: Add Tez AM JobID to the JobConf
Mar 13, 2024
452241a
Addressing PR comments
Mar 18, 2024
f9e3b45
Addressing PR comments #2
Apr 1, 2024
e13d6f4
Rename UUID property
Apr 5, 2024
5cef95a
Refactor createJobUUID to getDAGID
Apr 8, 2024
83cd25f
Address review comments
May 28, 2024
c0c29b4
Refactor to address more review comments
Jun 26, 2024
c56e926
Address checkstyle report
Jul 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions tez-api/src/main/java/org/apache/tez/runtime/api/OutputCommitterContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,4 +78,6 @@ public interface OutputCommitterContext {
*/
public int getVertexIndex();

public int getDagIdentifier();

}
10 changes: 9 additions & 1 deletion tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/OutputCommitterContextImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,14 +34,16 @@ public class OutputCommitterContextImpl implements OutputCommitterContext {
private final String dagName;
private final String vertexName;
private final int vertexIdx;
private final int dagIdentifier;
private final RootInputLeafOutput<OutputDescriptor, OutputCommitterDescriptor> output;

public OutputCommitterContextImpl(ApplicationId applicationId,
int dagAttemptNumber,
String dagName,
String vertexName,
RootInputLeafOutput<OutputDescriptor, OutputCommitterDescriptor> output,
int vertexIdx) {
int vertexIdx,
int dagIdentifier) {
Objects.requireNonNull(applicationId, "applicationId is null");
Objects.requireNonNull(dagName, "dagName is null");
Objects.requireNonNull(vertexName, "vertexName is null");
Expand All @@ -52,6 +54,7 @@ public OutputCommitterContextImpl(ApplicationId applicationId,
this.vertexName = vertexName;
this.output = output;
this.vertexIdx = vertexIdx;
this.dagIdentifier = dagIdentifier;
}

@Override
Expand Down Expand Up @@ -94,4 +97,9 @@ public int getVertexIndex() {
return vertexIdx;
}

@Override
public int getDagIdentifier() {
return dagIdentifier;
}

}
3 changes: 2 additions & 1 deletion tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2561,7 +2561,8 @@ public Void run() throws Exception {
appContext.getCurrentDAG().getName(),
vertexName,
od,
vertexId.getId());
vertexId.getId(),
appContext.getCurrentDAG().getID().getId());
OutputCommitter outputCommitter = ReflectionUtils
.createClazzInstance(od.getControllerDescriptor().getClassName(),
new Class[]{OutputCommitterContext.class},
Expand Down
4 changes: 4 additions & 0 deletions tez-mapreduce/src/main/java/org/apache/tez/mapreduce/committer/MROutputCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

package org.apache.tez.mapreduce.committer;

import org.apache.tez.mapreduce.common.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.classification.InterfaceAudience.Public;
Expand Down Expand Up @@ -78,6 +79,9 @@ public void initialize() throws IOException {
jobConf.getCredentials().mergeAll(UserGroupInformation.getCurrentUser().getCredentials());
jobConf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID,
getContext().getDAGAttemptNumber());
jobConf.set(MRJobConfig.FS_S3A_COMMITTER_UUID, Utils.createJobUUID(
getContext().getApplicationId().getClusterTimestamp(),
getContext().getApplicationId().getId(), getContext().getDagIdentifier()));
jobConf.setInt(MRJobConfig.VERTEX_ID, getContext().getVertexIndex());
committer = getOutputCommitter(getContext());
jobContext = getJobContextFromVertexContext(getContext());
Expand Down
6 changes: 5 additions & 1 deletion tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem.Statistics;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapred.JobID;
import org.apache.tez.common.counters.TezCounter;
import org.apache.tez.mapreduce.hadoop.mapred.MRCounters;

Expand Down Expand Up @@ -63,5 +64,8 @@ public static Counter getMRCounter(TezCounter tezCounter) {
Objects.requireNonNull(tezCounter);
return new MRCounters.MRCounter(tezCounter);
}


public static String createJobUUID(long clusterId, int appId, int dagIdentifier) {
return new JobID(String.valueOf(clusterId), appId).toString() + "_" + String.valueOf(dagIdentifier);
}
}
5 changes: 5 additions & 0 deletions tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRJobConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,11 @@ public interface MRJobConfig {

public static final String CACHE_ARCHIVES_VISIBILITIES = "mapreduce.job.cache.archives.visibilities";

/**
* Can be used by downstream applications to set a DAG-wide UUID for some committers which need one
*/
public static final String FS_S3A_COMMITTER_UUID = "fs.s3a.committer.uuid";

public static final String FILEOUTPUTCOMMITTER_ALGORITHM_VERSION = "mapreduce.fileoutputcommitter.algorithm.version";

/**
Expand Down
3 changes: 3 additions & 0 deletions tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import org.apache.tez.common.Preconditions;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.tez.mapreduce.common.Utils;
import org.apache.tez.runtime.library.api.IOInterruptedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -413,6 +414,8 @@ protected List<Event> initializeBase() throws IOException, InterruptedException
}
jobConf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID,
getContext().getDAGAttemptNumber());
jobConf.set(MRJobConfig.FS_S3A_COMMITTER_UUID, Utils.createJobUUID(getContext().getApplicationId().getClusterTimestamp(),
getContext().getApplicationId().getId(), getContext().getDagIdentifier()));
TaskAttemptID taskAttemptId = org.apache.tez.mapreduce.hadoop.mapreduce.TaskAttemptContextImpl
.createMockTaskAttemptID(getContext().getApplicationId().getClusterTimestamp(),
getContext().getTaskVertexIndex(), getContext().getApplicationId().getId(),
Expand Down
22 changes: 22 additions & 0 deletions tez-mapreduce/src/test/java/org/apache/tez/mapreduce/output/TestMROutput.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,9 @@
import org.apache.tez.hadoop.shim.DefaultHadoopShim;
import org.apache.tez.mapreduce.TestUmbilical;
import org.apache.tez.mapreduce.TezTestUtils;
import org.apache.tez.mapreduce.common.Utils;
import org.apache.tez.mapreduce.hadoop.MRConfig;
import org.apache.tez.mapreduce.hadoop.MRJobConfig;
import org.apache.tez.runtime.LogicalIOProcessorRuntimeTask;
import org.apache.tez.runtime.api.OutputContext;
import org.apache.tez.runtime.api.ProcessorContext;
Expand Down Expand Up @@ -131,6 +133,26 @@ public void testMergeConfig() throws Exception {
assertEquals("base-value", mergedConf.get("base-key"));
}

@Test
public void testJobUUIDSet() throws Exception {
Configuration conf = new Configuration();
conf.setBoolean(MRConfig.IS_MAP_PROCESSOR, true);
DataSinkDescriptor dataSink = MROutput
.createConfigBuilder(conf, TextOutputFormat.class,
tmpDir.getPath())
.build();

OutputContext outputContext = createMockOutputContext(dataSink.getOutputDescriptor().getUserPayload(),
new Configuration(false));
MROutput output = new MROutput(outputContext, 2);
output.initialize();
String invalidJobID = "invalid default";
String parentJobID = output.jobConf.get(MRJobConfig.FS_S3A_COMMITTER_UUID, invalidJobID);
assertNotEquals(parentJobID, invalidJobID);
assertNotEquals(output.jobConf.get(org.apache.hadoop.mapred.JobContext.TASK_ATTEMPT_ID), parentJobID);
assertEquals(parentJobID, Utils.createJobUUID(outputContext.getApplicationId().getClusterTimestamp(), outputContext.getApplicationId().getId(), outputContext.getDagIdentifier()));
}

@Test(timeout = 5000)
public void testOldAPI_TextOutputFormat() throws Exception {
Configuration conf = new Configuration();
Expand Down