HADOOP-17833. Improve Magic Committer performance (apache#3289) (apache#4470)

Speed up the magic committer with key changes being

* Writes under __magic always retain directory markers

* File creation under __magic skips all overwrite checks,
  including the LIST call intended to stop files being
        created over dirs.
* mkdirs under __magic probes the path for existence
  but does not look any further.

Extra parallelism in task and job commit directory scanning
Use of createFile and openFile with parameters which all for
HEAD checks to be skipped.

The committer can write the summary _SUCCESS file to the path
`fs.s3a.committer.summary.report.directory`, which can be in a
different file system/bucket if desired, using the job id as
the filename.

Also: HADOOP-15460. S3A FS to add `fs.s3a.create.performance`

Application code can set the createFile() option
fs.s3a.create.performance to true to disable the same
safety checks when writing under magic directories.
Use with care.

The createFile option prefix `fs.s3a.create.header.`
can be used to add custom headers to S3 objects when
created.

Contributed by Steve Loughran.

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/audit/AuditConstants.java

@@ -90,6 +90,11 @@ private AuditConstants() {
    */
   public static final String PARAM_PROCESS = "ps";
 
+  /**
+   * Task Attempt ID query header: {@value}.
+   */
+  public static final String PARAM_TASK_ATTEMPT_ID = "ta";
+
   /**
    * Thread 0: the thread which created a span {@value}.
    */

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/audit/CommonAuditContext.java

@@ -116,11 +116,15 @@ private CommonAuditContext() {
   /**
    * Put a context entry.
    * @param key key
-   * @param value new value
+   * @param value new value., If null, triggers removal.
    * @return old value or null
    */
   public Supplier<String> put(String key, String value) {
-    return evaluatedEntries.put(key, () -> value);
+    if (value != null) {
+      return evaluatedEntries.put(key, () -> value);
+    } else {
+      return evaluatedEntries.remove(key);
+    }
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/statistics/StoreStatisticNames.java

@@ -47,6 +47,9 @@ public final class StoreStatisticNames {
   /** {@value}. */
   public static final String OP_CREATE = "op_create";
 
+  /** {@value}. */
+  public static final String OP_CREATE_FILE = "op_createfile";
+
   /** {@value}. */
   public static final String OP_CREATE_NON_RECURSIVE =
       "op_create_non_recursive";

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/JsonSerialization.java

@@ -270,11 +270,12 @@ public void save(FileSystem fs, Path path, T instance,
   }
 
   /**
-   * Write the JSON as bytes, then close the file.
+   * Write the JSON as bytes, then close the stream.
+   * @param instance instance to write
    * @param dataOutputStream an output stream that will always be closed
    * @throws IOException on any failure
    */
-  private void writeJsonAsBytes(T instance,
+  public void writeJsonAsBytes(T instance,
       OutputStream dataOutputStream) throws IOException {
     try {
       dataOutputStream.write(toBytes(instance));

hadoop-common-project/hadoop-common/src/site/markdown/filesystem/fsdataoutputstreambuilder.md

@@ -26,7 +26,7 @@ create a new file or open an existing file on `FileSystem` for write.
 ## Invariants
 
 The `FSDataOutputStreamBuilder` interface does not validate parameters
-and modify the state of `FileSystem` until [`build()`](#Builder.build) is
+and modify the state of `FileSystem` until `build()` is
 invoked.
 
 ## Implementation-agnostic parameters.
@@ -110,7 +110,7 @@ of `FileSystem`.
 #### Implementation Notes
 
 The concrete `FileSystem` and/or `FSDataOutputStreamBuilder` implementation
-MUST verify that implementation-agnostic parameters (i.e., "syncable") or
+MUST verify that implementation-agnostic parameters (i.e., "syncable`) or
 implementation-specific parameters (i.e., "foofs:cache")
 are supported. `FileSystem` will satisfy optional parameters (via `opt(key, ...)`)
 on best effort. If the mandatory parameters (via `must(key, ...)`) can not be satisfied
@@ -182,3 +182,58 @@ see `FileSystem#create(path, ...)` and `FileSystem#append()`.
     result = FSDataOutputStream
 
 The result is `FSDataOutputStream` to be used to write data to filesystem.
+
+
+## <a name="s3a"></a> S3A-specific options
+
+Here are the custom options which the S3A Connector supports.
+
+| Name                        | Type      | Meaning                                |
+|-----------------------------|-----------|----------------------------------------|
+| `fs.s3a.create.performance` | `boolean` | create a file with maximum performance |
+| `fs.s3a.create.header`      | `string`  | prefix for user supplied headers       |
+
+### `fs.s3a.create.performance`
+
+Prioritize file creation performance over safety checks for filesystem consistency.
+
+This:
+1. Skips the `LIST` call which makes sure a file is being created over a directory.
+   Risk: a file is created over a directory.
+1. Ignores the overwrite flag.
+1. Never issues a `DELETE` call to delete parent directory markers.
+
+It is possible to probe an S3A Filesystem instance for this capability through
+the `hasPathCapability(path, "fs.s3a.create.performance")` check.
+
+Creating files with this option over existing directories is likely
+to make S3A filesystem clients behave inconsistently.
+
+Operations optimized for directories (e.g. listing calls) are likely
+to see the directory tree not the file; operations optimized for
+files (`getFileStatus()`, `isFile()`) more likely to see the file.
+The exact form of the inconsistencies, and which operations/parameters
+trigger this are undefined and may change between even minor releases.
+
+Using this option is the equivalent of pressing and holding down the
+"Electronic Stability Control"
+button on a rear-wheel drive car for five seconds: the safety checks are off.
+Things wil be faster if the driver knew what they were doing.
+If they didn't, the fact they had held the button down will
+be used as evidence at the inquest as proof that they made a
+conscious decision to choose speed over safety and
+that the outcome was their own fault.
+
+Accordingly: *Use if and only if you are confident that the conditions are met.*
+
+### `fs.s3a.create.header` User-supplied header support
+
+Options with the prefix `fs.s3a.create.header.` will be added to to the
+S3 object metadata as "user defined metadata".
+This metadata is visible to all applications. It can also be retrieved through the
+FileSystem/FileContext `listXAttrs()` and `getXAttrs()` API calls with the prefix `header.`
+
+When an object is renamed, the metadata is propagated the copy created.
+
+It is possible to probe an S3A Filesystem instance for this capability through
+the `hasPathCapability(path, "fs.s3a.create.header")` check.

hadoop-tools/hadoop-aws/pom.xml

@@ -396,6 +396,39 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-enforcer-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>banned-illegal-imports</id>
+            <phase>process-sources</phase>
+            <goals>
+              <goal>enforce</goal>
+            </goals>
+            <configuration>
+              <rules>
+                <restrictImports>
+                  <includeTestCode>false</includeTestCode>
+                  <reason>Restrict mapreduce imports to committer code</reason>
+                  <exclusions>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.AbstractS3ACommitter</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.AbstractS3ACommitterFactory</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.impl.*</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.magic.*</exclusion>
+                    <exclusion>org.apache.hadoop.fs.s3a.commit.staging.*</exclusion>
+                  </exclusions>
+                  <bannedImports>
+                    <bannedImport>org.apache.hadoop.mapreduce.**</bannedImport>
+                    <bannedImport>org.apache.hadoop.mapred.**</bannedImport>
+                  </bannedImports>
+                </restrictImports>
+              </rules>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/api/RequestFactory.java

@@ -49,6 +49,7 @@
 import org.apache.hadoop.fs.PathIOException;
 import org.apache.hadoop.fs.s3a.S3AEncryptionMethods;
 import org.apache.hadoop.fs.s3a.auth.delegation.EncryptionSecrets;
+import org.apache.hadoop.fs.s3a.impl.PutObjectOptions;
 
 /**
  * Factory for S3 objects.
@@ -128,23 +129,26 @@ CopyObjectRequest newCopyObjectRequest(String srcKey,
    * Adds the ACL and metadata
    * @param key key of object
    * @param metadata metadata header
+   * @param options options for the request
    * @param srcfile source file
    * @return the request
    */
   PutObjectRequest newPutObjectRequest(String key,
-      ObjectMetadata metadata, File srcfile);
+      ObjectMetadata metadata, PutObjectOptions options, File srcfile);
 
   /**
    * Create a {@link PutObjectRequest} request.
    * The metadata is assumed to have been configured with the size of the
    * operation.
    * @param key key of object
    * @param metadata metadata header
+   * @param options options for the request
    * @param inputStream source data.
    * @return the request
    */
   PutObjectRequest newPutObjectRequest(String key,
       ObjectMetadata metadata,
+      PutObjectOptions options,
       InputStream inputStream);
 
   /**
@@ -177,10 +181,12 @@ AbortMultipartUploadRequest newAbortMultipartUploadRequest(
   /**
    * Start a multipart upload.
    * @param destKey destination object key
+   * @param options options for the request
    * @return the request.
    */
   InitiateMultipartUploadRequest newMultipartUploadRequest(
-      String destKey);
+      String destKey,
+      @Nullable PutObjectOptions options);
 
   /**
    * Complete a multipart upload.