class org.apache.hadoop.fs.FutureDataInputStreamBuilder

Hadoop 3.3.0: API introduced

HADOOP-15229 Add FileSystem builder-based openFile() API to match createFile()

• No opt(String key, long value) method was available.
• the withFileStatus(status) call required a non-null parameter.
• Sole Filesystem to process options and file status was S3A;
• Only the s3a specific options were the S3 select and fs.s3a.experimental.input.fadvise
• S3A Filesystem raised IllegalArgumentException if a file status was passed in and the path of the filestatus did not match the path of the openFile(path) call.

This is the baseline implementation. To write code guaranteed to compile against this version, use the opt(String, String) and must(String, String) methods, converting numbers to string explicitly.

fs.open("s3a://bucket/file")
  .opt("fs.option.openfile.length", Long.toString(length))
  .build().get()

Hadoop 3.3.5: standardization and expansion

HADOOP-16202 Enhance openFile() for better read performance against object stores

• withFileStatus(null) required to be accepted (and ignored)
• only the filename part of any supplied FileStatus path must match the filename passed in on openFile(path).
• An opt(String key, long value) option was added. *This is now deprecated as it caused regression
• Standard fs.option.openfile options defined.
• S3A FS to use openfile length option, seek start/end options not yet used.
• Azure ABFS connector takes a supplied VersionedFileStatus and omits any HEAD probe for the object.

Hadoop 3.3.6: API change to address operator overload bugs.

new optLong(), optDouble(), mustLong() and mustDouble() builder methods.

• See HADOOP-18724 Open file fails with NumberFormatException for S3AFileSystem, which was somehow caused by the overloaded opt(long).
• Specification updated to declare that unparseable numbers MUST be treated as “unset” and the default value used instead.

FutureDataInputStreamBuilder bufferSize(int bufSize)

Set the size of the buffer to be used.

FutureDataInputStreamBuilder withFileStatus(FileStatus status)

A FileStatus instance which refers to the file being opened.

This MAY be used by implementations to short-circuit checks for the file, So potentially saving on remote calls especially to object stores.

Requirements:

• status != null
• status.getPath().getName() == the name of the file being opened.

The path validation MUST take place if the store uses the FileStatus when it opens files, and MAY be performed otherwise. The validation SHOULD be postponed until the build() operation.

This operation should be considered a hint to the filesystem.

If a filesystem implementation extends the FileStatus returned in its implementation MAY use this information when opening the file.

This is relevant with those stores which return version/etag information, -they MAY use this to guarantee that the file they opened is exactly the one returned in the listing.

The final status.getPath().getName() element of the supplied status MUST equal the name value of the path supplied to the openFile(path) call.

Filesystems MUST NOT validate the rest of the path. This is needed to support viewfs and other mount-point wrapper filesystems where schemas and paths are different. These often create their own FileStatus results

Preconditions

status == null or status.getPath().getName() == path.getName()

Filesystems MUST NOT require the class of status to equal that of any specific subclass their implementation returns in filestatus/list operations. This is to support wrapper filesystems and serialization/deserialization of the status.

Set optional or mandatory parameters

FutureDataInputStreamBuilder opt(String key, String value)
FutureDataInputStreamBuilder opt(String key, int value)
FutureDataInputStreamBuilder opt(String key, boolean value)
FutureDataInputStreamBuilder optLong(String key, long value)
FutureDataInputStreamBuilder optDouble(String key, double value)
FutureDataInputStreamBuilder must(String key, String value)
FutureDataInputStreamBuilder must(String key, int value)
FutureDataInputStreamBuilder must(String key, boolean value)
FutureDataInputStreamBuilder mustLong(String key, long value)
FutureDataInputStreamBuilder mustDouble(String key, double value)

Set optional or mandatory parameters to the builder. Using opt() or must(), client can specify FS-specific parameters without inspecting the concrete type of FileSystem.

Example:

out = fs.openFile(path)
    .must("fs.option.openfile.read.policy", "random")
    .optLong("fs.http.connection.timeout", 30_000L)
    .withFileStatus(statusFromListing)
    .build()
    .get();

Here the read policy of random has been specified, with the requirement that the filesystem implementation must understand the option. An http-specific option has been supplied which may be interpreted by any store; If the filesystem opening the file does not recognize the option, it can safely be ignored.

When to use opt versus must

The difference between opt versus must is how the FileSystem opening the file must react to an option which it does not recognize.

def must(name, value):
  if not name in known_keys:
    raise IllegalArgumentException
  if not name in supported_keys:
    raise UnsupportedException


def opt(name, value):
  if not name in known_keys:
     # ignore option

For any known key, the validation of the value argument MUST be the same irrespective of how the (key, value) pair was declared.

1. For a filesystem-specific option, it is the choice of the implementation how to validate the entry.
2. For standard options, the specification of what is a valid value is defined in this filesystem specification, validated through contract tests.

CompletableFuture<FSDataInputStream> build()

Return an CompletableFuture<FSDataInputStream> which, when successfully completed, returns an input stream which can read data from the filesystem.

The build() operation MAY perform the validation of the file’s existence, its kind, so rejecting attempts to read from a directory or non-existent file. Alternatively * file existence/status checks MAY be performed asynchronously within the returned CompletableFuture<>. * file existence/status checks MAY be postponed until the first byte is read in any of the read such as read() or PositionedRead.

That is, the precondition exists(FS, path) and isFile(FS, path) are only guaranteed to have been met after the get() called on returned future and an attempt has been made to read the stream.

Thus, if even when file does not exist, or is a directory rather than a file, the following call MUST succeed, returning a CompletableFuture to be evaluated.

Path p = new Path("file://tmp/file-which-does-not-exist");

CompletableFuture<FSDataInputStream> future = p.getFileSystem(conf)
      .openFile(p)
      .build();

The inability to access/read a file MUST raise an IOExceptionor subclass in either the future’s get() call, or, for late binding operations, when an operation to read data is invoked.

Therefore the following sequence SHALL fail when invoked on the future returned by the previous example.

  future.get().read();

Access permission checks have the same visibility requirements: permission failures MUST be delayed until the get() call and MAY be delayed into subsequent operations.

Note: some operations on the input stream, such as seek() may not attempt any IO at all. Such operations MAY NOT raise exceotions when interacting with nonexistent/unreadable files.

Option: fs.option.openfile.buffer.size

Read buffer size in bytes.

This overrides the default value set in the configuration with the option io.file.buffer.size.

It is supported by all filesystem clients which allow for stream-specific buffer sizes to be set via FileSystem.open(path, buffersize).

Option: fs.option.openfile.read.policy

Declare the read policy of the input stream. This is a hint as to what the expected read pattern of an input stream will be. This MAY control readahead, buffering and other optimizations.

Sequential reads may be optimized with prefetching data and/or reading data in larger blocks. Some applications (e.g. distCp) perform sequential IO even over columnar data.

In contrast, random IO reads data in different parts of the file using a sequence of seek()/read() or via the PositionedReadable or ByteBufferPositionedReadable APIs.

Random IO performance may be best if little/no prefetching takes place, along with other possible optimizations

Queries over columnar formats such as Apache ORC and Apache Parquet perform such random IO; other data formats may be best read with sequential or whole-file policies.

What is key is that optimizing reads for seqential reads may impair random performance -and vice versa.

1. The seek policy is a hint; even if declared as a must() option, the filesystem MAY ignore it.
2. The interpretation/implementation of a policy is a filesystem specific behavior -and it may change with Hadoop releases and/or specific storage subsystems.
3. If a policy is not recognized, the filesystem client MUST ignore it.

Choosing the wrong read policy for an input source may be inefficient.

A list of read policies MAY be supplied; the first one recognized/supported by the filesystem SHALL be the one used. This allows for custom policies to be supported, for example an hbase-hfile policy optimized for HBase HFiles.

The S3A and ABFS input streams both implement the IOStatisticsSource API, and can be queried for their IO Performance.

Tip: log the toString() value of input streams at DEBUG. The S3A and ABFS Input Streams log read statistics, which can provide insight about whether reads are being performed efficiently or not.

Futher reading

• Linux fadvise().
• Windows CreateFile()

Option: fs.option.openfile.length

Declare the length of a file.

This can be used by clients to skip querying a remote store for the size of/existence of a file when opening it, similar to declaring a file status through the withFileStatus() option.

If supported by a filesystem connector, this option MUST be interpreted as declaring the minimum length of the file:

1. If the value is negative, the option SHALL be considered unset.
2. It SHALL NOT be an error if the actual length of the file is greater than this value.
3. read(), seek() and positioned read calls MAY use a position across/beyond this length but below the actual length of the file. Implementations MAY raise EOFExceptions in such cases, or they MAY return data.

If this option is used by the FileSystem implementation

Implementor’s Notes

• A value of fs.option.openfile.length < 0 MUST be ignored.
• If a file status is supplied along with a value in fs.opt.openfile.length; the file status values take precedence.

Options: fs.option.openfile.split.start and fs.option.openfile.split.end

Declare the start and end of the split when a file has been split for processing in pieces.

1. If a value is negative, the option SHALL be considered unset.
2. Filesystems MAY assume that the length of the file is greater than or equal to the value of fs.option.openfile.split.end.
3. And that they MAY raise an exception if the client application reads past the value set in fs.option.openfile.split.end.
4. The pair of options MAY be used to optimise the read plan, such as setting the content range for GET requests, or using the split end as an implicit declaration of the guaranteed minimum length of the file.
5. If both options are set, and the split start is declared as greater than the split end, then the split start SHOULD just be reset to zero, rather than rejecting the operation.

The split end value can provide a hint as to the end of the input stream. The split start can be used to optimize any initial read offset for filesystem clients.

*Note for implementors: applications will read past the end of a split when they need to read to the end of a record/line which begins before the end of the split.

Therefore clients MUST be allowed to seek()/read() past the length set in fs.option.openfile.split.end if the file is actually longer than that value.

Opening a file with both standard and non-standard options

Standard and non-standard options MAY be combined in the same openFile() operation.

Future<FSDataInputStream> f = openFile(path)
  .must("fs.option.openfile.read.policy", "random, adaptive")
  .opt("fs.s3a.readahead.range", 1024 * 1024)
  .build();

FSDataInputStream is = f.get();

The option set in must() MUST be understood, or at least recognized and ignored by all filesystems. In this example, S3A-specific option MAY be ignored by all other filesystem clients.

Opening a file with older releases

Not all hadoop releases recognize the fs.option.openfile.read.policy option.

The option can be safely used in application code if it is added via the opt() builder argument, as it will be treated as an unknown optional key which can then be discarded.

Future<FSDataInputStream> f = openFile(path)
  .opt("fs.option.openfile.read.policy", "vector, random, adaptive")
  .build();

FSDataInputStream is = f.get();

Note 1 if the option name is set by a reference to a constant in org.apache.hadoop.fs.Options.OpenFileOptions, then the program will not link against versions of Hadoop without the specific option. Therefore for resilient linking against older releases -use a copy of the value.

Note 2 as option validation is performed in the FileSystem connector, a third-party connector designed to work with multiple hadoop versions MAY NOT support the option.

Passing options in to MapReduce

Hadoop MapReduce will automatically read MR Job Options with the prefixes mapreduce.job.input.file.option. and mapreduce.job.input.file.must. prefixes, and apply these values as .opt() and must() respectively, after remove the mapreduce-specific prefixes.

This makes passing options in to MR jobs straightforward. For example, to declare that a job should read its data using random IO:

JobConf jobConf = (JobConf) job.getConfiguration()
jobConf.set(
    "mapreduce.job.input.file.option.fs.option.openfile.read.policy",
    "random");

MapReduce input format propagating options

An example of a record reader passing in options to the file it opens.

  public void initialize(InputSplit genericSplit,
                     TaskAttemptContext context) throws IOException {
    FileSplit split = (FileSplit)genericSplit;
    Configuration job = context.getConfiguration();
    start = split.getStart();
    end = start + split.getLength();
    Path file = split.getPath();

    // open the file and seek to the start of the split
    FutureDataInputStreamBuilder builder =
      file.getFileSystem(job).openFile(file);
    // the start and end of the split may be used to build
    // an input strategy.
    builder.optLong("fs.option.openfile.split.start", start);
    builder.optLong("fs.option.openfile.split.end", end);
    FutureIO.propagateOptions(builder, job,
        "mapreduce.job.input.file.option",
        "mapreduce.job.input.file.must");

    fileIn = FutureIO.awaitFuture(builder.build());
    fileIn.seek(start)
    /* Rest of the operation on the opened stream */
  }

FileContext.openFile

From org.apache.hadoop.fs.AvroFSInput; a file is opened with sequential input. Because the file length has already been probed for, the length is passed down

  public AvroFSInput(FileContext fc, Path p) throws IOException {
    FileStatus status = fc.getFileStatus(p);
    this.len = status.getLen();
    this.stream = awaitFuture(fc.openFile(p)
        .opt("fs.option.openfile.read.policy",
            "sequential")
        .optLong("fs.option.openfile.length",
            Long.toString(status.getLen()))
        .build());
    fc.open(p);
  }

In this example, the length is passed down as a string (via Long.toString()) rather than directly as a long. This is to ensure that the input format will link against versions of $Hadoop which do not have the opt(String, long) and must(String, long) builder parameters. Similarly, the values are passed as optional, so that if unrecognized the application will still succeed.

Example: reading a whole file

This is from org.apache.hadoop.util.JsonSerialization.

Its load(FileSystem, Path, FileStatus) method * declares the whole file is to be read end to end. * passes down the file status

public T load(FileSystem fs,
        Path path,
        status)
        throws IOException {

 try (FSDataInputStream dataInputStream =
          awaitFuture(fs.openFile(path)
              .opt("fs.option.openfile.read.policy", "whole-file")
              .withFileStatus(status)
              .build())) {
   return fromJsonStream(dataInputStream);
 } catch (JsonProcessingException e) {
   throw new PathIOException(path.toString(),
       "Failed to read JSON file " + e, e);
 }
}