TEZ-4547: Add Tez AM JobID to the JobConf #339
Conversation
Some committers require a job-wide UUID to function correctly. Adding the AM JobID to the JobConf will allow applications to pass that to the committers that need it.
VenkatSNarayanan
force-pushed
the
vnarayanan/TEZ-4547
branch
from
25cae6d
to
9040a2f
Compare
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
| @@ -417,6 +418,7 @@ protected List<Event> initializeBase() throws IOException, InterruptedException | |||
| .createMockTaskAttemptID(getContext().getApplicationId().getClusterTimestamp(), | |||
| getContext().getTaskVertexIndex(), getContext().getApplicationId().getId(), | |||
| getContext().getTaskIndex(), getContext().getTaskAttemptNumber(), isMapperOutput); | |||
| jobConf.set(MRJobConfig.MR_PARENT_JOB_ID, new JobID(String.valueOf(getContext().getApplicationId().getClusterTimestamp()), getContext().getApplicationId().getId()).toString()); | |||
There was a problem hiding this comment.
Move this to line above TaskAttemptID taskAttemptId =
| assertNotEquals(parentJobID,invalidJobID); | ||
| assertNotEquals(output.jobConf.get(org.apache.hadoop.mapred.JobContext.TASK_ATTEMPT_ID),parentJobID); |
There was a problem hiding this comment.
Fix code formatting. space after ,
| @@ -119,6 +119,7 @@ public void abortOutput(VertexStatus.State finalState) throws IOException { | |||
| || jobConf.getBoolean("mapred.mapper.new-api", false)) { | |||
| newApiCommitter = true; | |||
| } | |||
| jobConf.set(MRJobConfig.MR_PARENT_JOB_ID, new org.apache.hadoop.mapred.JobID(String.valueOf(getContext().getApplicationId().getClusterTimestamp()), getContext().getApplicationId().getId()).toString()); | |||
There was a problem hiding this comment.
Can we have String.valueOf(getContext().getApplicationId().getClusterTimestamp()), getContext().getApplicationId().getId()).toString( inside a method and re-use it in MROutput.java as well ?
Yes, that was the plan. The property name was just chosen arbitrarily so I could put the PR up, any suggestions for a better one are welcome. |
This commit also adds the DAG identifier to the job UUID to ensure that multiple jobs within the same session will be assigned different UUIDs.
This comment was marked as outdated.
This comment was marked as outdated.
tez-mapreduce/src/main/java/org/apache/tez/mapreduce/committer/MROutputCommitter.java
Outdated
Show resolved
Hide resolved
tez-mapreduce/src/main/java/org/apache/tez/mapreduce/common/Utils.java
Outdated
Show resolved
Hide resolved
tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRJobConfig.java
Outdated
Show resolved
Hide resolved
tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java
Outdated
Show resolved
Hide resolved
This comment was marked as outdated.
This comment was marked as outdated.
|
@shameersss1 We could actually just set fs.s3a.committer.uuid directly instead of the indirection through the other setting. |
Switch UUID property name to the one required by S3A committers.
This comment was marked as outdated.
This comment was marked as outdated.
|
@abstractdog - Could you please review the same ? |
Refactors the implementation to reuse Tez's DAGID type instead of hand-rolling our own.
This comment was marked as outdated.
This comment was marked as outdated.
|
@abstractdog @shameersss1 Is there anything else needed? |
tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRJobConfig.java
Outdated
Show resolved
Hide resolved
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/OutputCommitterContextImpl.java
Outdated
Show resolved
Hide resolved
This comment was marked as outdated.
This comment was marked as outdated.
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/OutputCommitterContextImpl.java
Show resolved
Hide resolved
tez-mapreduce/src/main/java/org/apache/tez/mapreduce/committer/MROutputCommitter.java
Outdated
Show resolved
Hide resolved
|
left minor comments on this @VenkatSNarayanan , other than that, this looks good to me |
|
🎊 +1 overall
This message was automatically generated. |
|
one more thing @VenkatSNarayanan , please address checkstyle comments where applicable, thanks! |
|
💔 -1 overall
This message was automatically generated. |
| /** | ||
| * Used by committers to set a job-wide UUID. | ||
| */ | ||
| public static final String JOB_COMMITTER_UUID = "job.committer.uuid"; |
There was a problem hiding this comment.
This is not the setting used by s3 committer right? How will it work ?
There was a problem hiding this comment.
There is a corresponding change I have in my Hadoop code where it will consult this property similar to how it consults the property Spark sets for this purpose.
There was a problem hiding this comment.
so you can confirm this will work with job.committer.uuid, right?
can you link that point in hadoop code for later reference?
There was a problem hiding this comment.
@VenkatSNarayanan May i ask that if the hadoop s3 committer can work with Hive+Tez after this change?
IMO, s3/magic committer can avoid some operation like rename on s3, which can speed up/improve hive job.
There was a problem hiding this comment.
@abstractdog I haven't publicly posted the Hadoop PR yet, but the change I have is to check for this property around here: https://github.com/apache/hadoop/blob/51cb858cc8c23d873d4adfc21de5f2c1c22d346f/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/AbstractS3ACommitter.java#L1372 similar to how the Spark property is checked. I have tested these changes together already alongside my Hive implementation.
@zhangbutao There are some corresponding changes to Hadoop and Hive that also need to be merged which I have. Once all 3 PRs(Tez, Hadoop and Hive have been merged), then the magic committer will be usable with Hive.
There was a problem hiding this comment.
should this go into Tez 0.10.4? if so, it would be good to have it in 1-2 weeks, just FYI, regarding planning the hadoop change
There was a problem hiding this comment.
0.10.4 would be ideal. In that case, let me loop in the Hadoop folks to see if they have any strong opinions about this.
There was a problem hiding this comment.
@VenkatSNarayanan https://issues.apache.org/jira/browse/HIVE-16295 I found a old ticket about integrating s3a committer, and it seems that supporting this needs lots of Hive code change.
I am not sure if you have done similar change in Hive to support the MagicS3GuardCommitter.
Anyway, I think it is very good to support this committer in Hive&Tez. Look forward to your further work.
Thanks.
There was a problem hiding this comment.
https://issues.apache.org/jira/browse/HADOOP-19091 I just saw your Hadoop ticket, and Hive change patch is also there too. Maybe you need create a PR against Hive latest master branch once you have done preparatory work. :)
There was a problem hiding this comment.
There haven't been any objections from the Hadoop folks, I think it should be safe to go ahead with the patch as it is @abstractdog .
| @@ -78,6 +79,7 @@ public void initialize() throws IOException { | |||
| jobConf.getCredentials().mergeAll(UserGroupInformation.getCurrentUser().getCredentials()); | |||
| jobConf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, | |||
| getContext().getDAGAttemptNumber()); | |||
| jobConf.set(MRJobConfig.JOB_COMMITTER_UUID, Utils.getDAGID(getContext())); | |||
There was a problem hiding this comment.
this unique across all jobs which may be writing to a table, even from other processes?
There was a problem hiding this comment.
Yes. This ID is unique to a DAG + attempt number - so if we have some other job, it'll have a different application ID component, while if an attempt fails and the DAG retries, the attempt number will be different.
|
ok. if you look into _SUCCESS json from an s3a or the manifest committer, then the job id is one of the root attributes, as is the source there's a java definition of this in |
|
guys: @steveloughran , @VenkatSNarayanan : please let me know if this PR is fine to be merged to tez (from hadoop's point of view)? I'm about to start the release process of 0.10.4 soon |
FYI: I'm about to merge this tomorrow to have this in tez 0.10.4 |
Some committers require a job-wide UUID to function correctly. Adding the AM JobID to the JobConf
will allow applications to pass that to
the committers that need it.