class org.apache.hadoop.fs.FutureDataInputStreamBuilder

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, ...)
FutureDataInputStreamBuilder must(String key, ...)

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")
    .opt("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 rejected.
• 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.opt("fs.option.openfile.split.start", start);
    builder.opt("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 passd 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")
        .opt("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);
 }
}

Note: : in Hadoop 3.3.2 and earlier, the withFileStatus(status) call required a non-null parameter; this has since been relaxed. For maximum compatibility across versions, only invoke the method when the file status is known to be non-null.