[
https://issues.apache.org/jira/browse/HADOOP-19047?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17814234#comment-17814234
]
ASF GitHub Bot commented on HADOOP-19047:
-----------------------------------------
shameersss1 commented on code in PR #6468:
URL: https://github.com/apache/hadoop/pull/6468#discussion_r1477795947
##########
hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/MagicS3GuardCommitter.java:
##########
@@ -248,6 +236,80 @@ private PendingSet innerCommitTask(
return pendingSet;
}
+ /**
+ * Loads pending commits from either memory or from the remote store (S3)
based on the config.
+ * @param context TaskAttemptContext
+ * @return All pending commit data for the given TaskAttemptContext
+ * @throws IOException
+ * if there is an error trying to read the commit data
+ */
+ protected PendingSet loadPendingCommits(TaskAttemptContext context) throws
IOException {
+ PendingSet pendingSet = new PendingSet();
+ if (isTrackMagicCommitsInMemoryEnabled(context.getConfiguration())) {
+ // load from memory
+ List<SinglePendingCommit> pendingCommits =
loadPendingCommitsFromMemory(context);
+
+ for (SinglePendingCommit singleCommit : pendingCommits) {
+ // aggregate stats
+ pendingSet.getIOStatistics()
+ .aggregate(singleCommit.getIOStatistics());
+ // then clear so they aren't marshalled again.
+ singleCommit.getIOStatistics().clear();
+ }
+ pendingSet.setCommits(pendingCommits);
+ } else {
+ // Load from remote store
+ CommitOperations actions = getCommitOperations();
+ Path taskAttemptPath = getTaskAttemptPath(context);
+ try (CommitContext commitContext = initiateTaskOperation(context)) {
+ Pair<PendingSet, List<Pair<LocatedFileStatus, IOException>>> loaded =
+ actions.loadSinglePendingCommits(taskAttemptPath, true,
commitContext);
+ pendingSet = loaded.getKey();
+ List<Pair<LocatedFileStatus, IOException>> failures =
loaded.getValue();
+ if (!failures.isEmpty()) {
+ // At least one file failed to load
+ // revert all which did; report failure with first exception
+ LOG.error("At least one commit file could not be read: failing");
+ abortPendingUploads(commitContext, pendingSet.getCommits(), true);
+ throw failures.get(0).getValue();
+ }
+ }
+ }
+ return pendingSet;
+ }
+
+ private List<SinglePendingCommit>
loadPendingCommitsFromMemory(TaskAttemptContext context)
Review Comment:
ack.
> Support InMemory Tracking Of S3A Magic Commits
> ----------------------------------------------
>
> Key: HADOOP-19047
> URL: https://issues.apache.org/jira/browse/HADOOP-19047
> Project: Hadoop Common
> Issue Type: Improvement
> Components: fs/s3
> Reporter: Syed Shameerur Rahman
> Assignee: Syed Shameerur Rahman
> Priority: Major
> Labels: pull-request-available
>
> The following are the operations which happens within a Task when it uses S3A
> Magic Committer.
> *During closing of stream*
> 1. A 0-byte file with a same name of the original file is uploaded to S3
> using PUT operation. Refer
> [here|https://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/MagicCommitTracker.java#L152]
> for more information. This is done so that the downstream application like
> Spark could get the size of the file which is being written.
> 2. MultiPartUpload(MPU) metadata is uploaded to S3. Refer
> [here|https://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/MagicCommitTracker.java#L176]
> for more information.
> *During TaskCommit*
> 1. All the MPU metadata which the task wrote to S3 (There will be 'x' number
> of metadata file in S3 if a single task writes to 'x' files) are read and
> rewritten to S3 as a single metadata file. Refer
> [here|https://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/MagicS3GuardCommitter.java#L201]
> for more information
> Since these operations happens with the Task JVM, We could optimize as well
> as save cost by storing these information in memory when Task memory usage is
> not a constraint. Hence the proposal here is to introduce a new MagicCommit
> Tracker called "InMemoryMagicCommitTracker" which will store the
> 1. Metadata of MPU in memory till the Task is committed
> 2. Store the size of the file which can be used by the downstream application
> to get the file size before it is committed/visible to the output path.
> This optimization will save 2 PUT S3 calls, 1 LIST S3 call, and 1 GET S3 call
> given a Task writes only 1 file.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
