HADOOP-13208. S3A listFiles(recursive=true) to do a bulk listObjects instead of walking the pseudo-tree of directories. Contributed by Steve Loughran.
(cherry picked from commit 822d661b8fcc42bec6eea958d9fd02ef1aaa4b6c) Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/986162f9 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/986162f9 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/986162f9 Branch: refs/heads/branch-2 Commit: 986162f9776837b1343073d721f89be8f3362085 Parents: 2486c4c Author: Chris Nauroth <cnaur...@apache.org> Authored: Wed Aug 17 14:54:54 2016 -0700 Committer: Chris Nauroth <cnaur...@apache.org> Committed: Wed Aug 17 14:55:07 2016 -0700 ---------------------------------------------------------------------- .../java/org/apache/hadoop/fs/s3a/Listing.java | 594 +++++++++++++++++++ .../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 372 +++++++----- .../hadoop/fs/s3a/S3AInstrumentation.java | 1 + .../java/org/apache/hadoop/fs/s3a/S3AUtils.java | 12 + .../org/apache/hadoop/fs/s3a/Statistic.java | 2 + .../s3a/TestS3AContractGetFileStatus.java | 16 + .../org/apache/hadoop/fs/s3a/S3ATestUtils.java | 8 +- .../s3a/scale/TestS3ADirectoryPerformance.java | 58 +- 8 files changed, 897 insertions(+), 166 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java new file mode 100644 index 0000000..4120b20 --- /dev/null +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java @@ -0,0 +1,594 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.fs.s3a; + +import com.amazonaws.AmazonClientException; +import com.amazonaws.services.s3.model.ListObjectsRequest; +import com.amazonaws.services.s3.model.ObjectListing; +import com.amazonaws.services.s3.model.S3ObjectSummary; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.LocatedFileStatus; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.PathFilter; +import org.apache.hadoop.fs.RemoteIterator; +import org.slf4j.Logger; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.ListIterator; +import java.util.NoSuchElementException; + +import static org.apache.hadoop.fs.s3a.Constants.S3N_FOLDER_SUFFIX; +import static org.apache.hadoop.fs.s3a.S3AUtils.createFileStatus; +import static org.apache.hadoop.fs.s3a.S3AUtils.objectRepresentsDirectory; +import static org.apache.hadoop.fs.s3a.S3AUtils.stringify; +import static org.apache.hadoop.fs.s3a.S3AUtils.translateException; + +/** + * Place for the S3A listing classes; keeps all the small classes under control. + */ +public class Listing { + + private final S3AFileSystem owner; + private static final Logger LOG = S3AFileSystem.LOG; + + public Listing(S3AFileSystem owner) { + this.owner = owner; + } + + /** + * Create a FileStatus iterator against a path, with a given + * list object request. + * @param listPath path of the listing + * @param request initial request to make + * @param filter the filter on which paths to accept + * @param acceptor the class/predicate to decide which entries to accept + * in the listing based on the full file status. + * @return the iterator + * @throws IOException IO Problems + */ + FileStatusListingIterator createFileStatusListingIterator( + Path listPath, + ListObjectsRequest request, + PathFilter filter, + Listing.FileStatusAcceptor acceptor) throws IOException { + return new FileStatusListingIterator( + new ObjectListingIterator(listPath, request), + filter, + acceptor); + } + + /** + * Create a located status iterator over a file status iterator. + * @param statusIterator an iterator over the remote status entries + * @return a new remote iterator + */ + LocatedFileStatusIterator createLocatedFileStatusIterator( + RemoteIterator<FileStatus> statusIterator) { + return new LocatedFileStatusIterator(statusIterator); + } + + /** + * Interface to implement by the logic deciding whether to accept a summary + * entry or path as a valid file or directory. + */ + interface FileStatusAcceptor { + + /** + * Predicate to decide whether or not to accept a summary entry. + * @param keyPath qualified path to the entry + * @param summary summary entry + * @return true if the entry is accepted (i.e. that a status entry + * should be generated. + */ + boolean accept(Path keyPath, S3ObjectSummary summary); + + /** + * Predicate to decide whether or not to accept a prefix. + * @param keyPath qualified path to the entry + * @param commonPrefix the prefix + * @return true if the entry is accepted (i.e. that a status entry + * should be generated.) + */ + boolean accept(Path keyPath, String commonPrefix); + } + + /** + * A remote iterator which only iterates over a single `LocatedFileStatus` + * value. + * + * If the status value is null, the iterator declares that it has no data. + * This iterator is used to handle {@link listStatus()} calls where the path + * handed in refers to a file, not a directory: this is the iterator + * returned. + */ + static final class SingleStatusRemoteIterator + implements RemoteIterator<LocatedFileStatus> { + + /** + * The status to return; set to null after the first iteration. + */ + private LocatedFileStatus status; + + /** + * Constructor. + * @param status status value: may be null, in which case + * the iterator is empty. + */ + public SingleStatusRemoteIterator(LocatedFileStatus status) { + this.status = status; + } + + /** + * {@inheritDoc} + * @return true if there is a file status to return: this is always false + * for the second iteration, and may be false for the first. + * @throws IOException never + */ + @Override + public boolean hasNext() throws IOException { + return status != null; + } + + /** + * {@inheritDoc} + * @return the non-null status element passed in when the instance was + * constructed, if it ha not already been retrieved. + * @throws IOException never + * @throws NoSuchElementException if this is the second call, or it is + * the first call and a null {@link LocatedFileStatus} entry was passed + * to the constructor. + */ + @Override + public LocatedFileStatus next() throws IOException { + if (hasNext()) { + LocatedFileStatus s = this.status; + status = null; + return s; + } else { + throw new NoSuchElementException(); + } + } + } + + /** + * Wraps up object listing into a remote iterator which will ask for more + * listing data if needed. + * + * This is a complex operation, especially the process to determine + * if there are more entries remaining. If there are no more results + * remaining in the (filtered) results of the current listing request, then + * another request is made <i>and those results filtered</i> before the + * iterator can declare that there is more data available. + * + * The need to filter the results precludes the iterator from simply + * declaring that if the {@link S3AFileSystem.ObjectListingIterator#hasNext()} + * is true then there are more results. Instead the next batch of results must + * be retrieved and filtered. + * + * What does this mean? It means that remote requests to retrieve new + * batches of object listings are made in the {@link #hasNext()} call; + * the {@link #next()} call simply returns the filtered results of the last + * listing processed. However, do note that {@link #next()} calls + * {@link #hasNext()} during its operation. This is critical to ensure + * that a listing obtained through a sequence of {@link #next()} will + * complete with the same set of results as a classic + * {@code while(it.hasNext()} loop. + * + * Thread safety: None. + */ + class FileStatusListingIterator + implements RemoteIterator<FileStatus> { + + /** Source of objects. */ + private final ObjectListingIterator source; + /** Filter of paths from API call. */ + private final PathFilter filter; + /** Filter of entries from file status. */ + private final FileStatusAcceptor acceptor; + /** request batch size. */ + private int batchSize; + /** Iterator over the current set of results. */ + private ListIterator<FileStatus> statusBatchIterator; + + /** + * Create an iterator over file status entries. + * @param source the listing iterator from a listObjects call. + * @param filter the filter on which paths to accept + * @param acceptor the class/predicate to decide which entries to accept + * in the listing based on the full file status. + * @throws IOException IO Problems + */ + FileStatusListingIterator(ObjectListingIterator source, + PathFilter filter, + FileStatusAcceptor acceptor) throws IOException { + this.source = source; + this.filter = filter; + this.acceptor = acceptor; + // build the first set of results. This will not trigger any + // remote IO, assuming the source iterator is in its initial + // iteration + requestNextBatch(); + } + + /** + * Report whether or not there is new data available. + * If there is data in the local filtered list, return true. + * Else: request more data util that condition is met, or there + * is no more remote listing data. + * @return true if a call to {@link #next()} will succeed. + * @throws IOException + */ + @Override + public boolean hasNext() throws IOException { + return statusBatchIterator.hasNext() || requestNextBatch(); + } + + @Override + public FileStatus next() throws IOException { + if (!hasNext()) { + throw new NoSuchElementException(); + } + return statusBatchIterator.next(); + } + + /** + * Try to retrieve another batch. + * Note that for the initial batch, + * {@link S3AFileSystem.ObjectListingIterator} does not generate a request; + * it simply returns the initial set. + * + * @return true if a new batch was created. + * @throws IOException IO problems + */ + private boolean requestNextBatch() throws IOException { + // look for more object listing batches being available + while (source.hasNext()) { + // if available, retrieve it and build the next status + if (buildNextStatusBatch(source.next())) { + // this batch successfully generated entries matching the filters/ + // acceptors; declare that the request was successful + return true; + } else { + LOG.debug("All entries in batch were filtered...continuing"); + } + } + // if this code is reached, it means that all remaining + // object lists have been retrieved, and there are no new entries + // to return. + return false; + } + + /** + * Build the next status batch from a listing. + * @param objects the next object listing + * @return true if this added any entries after filtering + */ + private boolean buildNextStatusBatch(ObjectListing objects) { + // counters for debug logs + int added = 0, ignored = 0; + // list to fill in with results. Initial size will be list maximum. + List<FileStatus> stats = new ArrayList<>( + objects.getObjectSummaries().size() + + objects.getCommonPrefixes().size()); + // objects + for (S3ObjectSummary summary : objects.getObjectSummaries()) { + String key = summary.getKey(); + Path keyPath = owner.keyToQualifiedPath(key); + if (LOG.isDebugEnabled()) { + LOG.debug("{}: {}", keyPath, stringify(summary)); + } + // Skip over keys that are ourselves and old S3N _$folder$ files + if (acceptor.accept(keyPath, summary) && filter.accept(keyPath)) { + FileStatus status = createFileStatus(keyPath, summary, + owner.getDefaultBlockSize(keyPath)); + LOG.debug("Adding: {}", status); + stats.add(status); + added++; + } else { + LOG.debug("Ignoring: {}", keyPath); + ignored++; + } + } + + // prefixes: always directories + for (String prefix : objects.getCommonPrefixes()) { + Path keyPath = owner.keyToQualifiedPath(prefix); + if (acceptor.accept(keyPath, prefix) && filter.accept(keyPath)) { + FileStatus status = new S3AFileStatus(true, false, keyPath); + LOG.debug("Adding directory: {}", status); + added++; + stats.add(status); + } else { + LOG.debug("Ignoring directory: {}", keyPath); + ignored++; + } + } + + // finish up + batchSize = stats.size(); + statusBatchIterator = stats.listIterator(); + boolean hasNext = statusBatchIterator.hasNext(); + LOG.debug("Added {} entries; ignored {}; hasNext={}; hasMoreObjects={}", + added, ignored, hasNext, objects.isTruncated()); + return hasNext; + } + + /** + * Get the number of entries in the current batch. + * @return a number, possibly zero. + */ + public int getBatchSize() { + return batchSize; + } + } + + /** + * Wraps up AWS `ListObjects` requests in a remote iterator + * which will ask for more listing data if needed. + * + * That is: + * + * 1. The first invocation of the {@link #next()} call will return the results + * of the first request, the one created during the construction of the + * instance. + * + * 2. Second and later invocations will continue the ongoing listing, + * calling {@link #continueListObjects(ObjectListing)} to request the next + * batch of results. + * + * 3. The {@link #hasNext()} predicate returns true for the initial call, + * where {@link #next()} will return the initial results. It declares + * that it has future results iff the last executed request was truncated. + * + * Thread safety: none. + */ + class ObjectListingIterator implements RemoteIterator<ObjectListing> { + + /** The path listed. */ + private final Path listPath; + + /** The most recent listing results. */ + private ObjectListing objects; + + /** Indicator that this is the first listing. */ + private boolean firstListing = true; + + /** + * Count of how many listings have been requested + * (including initial result). + */ + private int listingCount = 1; + + /** + * Maximum keys in a request. + */ + private int maxKeys; + + /** + * Constructor -calls `listObjects()` on the request to populate the + * initial set of results/fail if there was a problem talking to the bucket. + * @param listPath path of the listing + * @param request initial request to make + * */ + ObjectListingIterator( + Path listPath, + ListObjectsRequest request) { + this.listPath = listPath; + this.maxKeys = owner.getMaxKeys(); + this.objects = owner.listObjects(request); + } + + /** + * Declare that the iterator has data if it is either is the initial + * iteration or it is a later one and the last listing obtained was + * incomplete. + * @throws IOException never: there is no IO in this operation. + */ + @Override + public boolean hasNext() throws IOException { + return firstListing || objects.isTruncated(); + } + + /** + * Ask for the next listing. + * For the first invocation, this returns the initial set, with no + * remote IO. For later requests, S3 will be queried, hence the calls + * may block or fail. + * @return the next object listing. + * @throws IOException if a query made of S3 fails. + * @throws NoSuchElementException if there is no more data to list. + */ + @Override + public ObjectListing next() throws IOException { + if (firstListing) { + // on the first listing, don't request more data. + // Instead just clear the firstListing flag so that it future calls + // will request new data. + firstListing = false; + } else { + try { + if (!objects.isTruncated()) { + // nothing more to request: fail. + throw new NoSuchElementException("No more results in listing of " + + listPath); + } + // need to request a new set of objects. + LOG.debug("[{}], Requesting next {} objects under {}", + listingCount, maxKeys, listPath); + objects = owner.continueListObjects(objects); + listingCount++; + LOG.debug("New listing status: {}", this); + } catch (AmazonClientException e) { + throw translateException("listObjects()", listPath, e); + } + } + return objects; + } + + @Override + public String toString() { + return "Object listing iterator against " + listPath + + "; listing count "+ listingCount + + "; isTruncated=" + objects.isTruncated(); + } + + /** + * Get the path listed. + * @return the path used in this listing. + */ + public Path getListPath() { + return listPath; + } + + /** + * Get the count of listing requests. + * @return the counter of requests made (including the initial lookup). + */ + public int getListingCount() { + return listingCount; + } + } + + /** + * Accept all entries except the base path and those which map to S3N + * pseudo directory markers. + */ + static class AcceptFilesOnly implements FileStatusAcceptor { + private final Path qualifiedPath; + + public AcceptFilesOnly(Path qualifiedPath) { + this.qualifiedPath = qualifiedPath; + } + + /** + * Reject a summary entry if the key path is the qualified Path, or + * it ends with {@code "_$folder$"}. + * @param keyPath key path of the entry + * @param summary summary entry + * @return true if the entry is accepted (i.e. that a status entry + * should be generated. + */ + @Override + public boolean accept(Path keyPath, S3ObjectSummary summary) { + return !keyPath.equals(qualifiedPath) + && !summary.getKey().endsWith(S3N_FOLDER_SUFFIX) + && !objectRepresentsDirectory(summary.getKey(), summary.getSize()); + } + + /** + * Accept no directory paths. + * @param keyPath qualified path to the entry + * @param prefix common prefix in listing. + * @return false, always. + */ + @Override + public boolean accept(Path keyPath, String prefix) { + return false; + } + } + + /** + * Take a remote iterator over a set of {@link FileStatus} instances and + * return a remote iterator of {@link LocatedFileStatus} instances. + */ + class LocatedFileStatusIterator + implements RemoteIterator<LocatedFileStatus> { + private final RemoteIterator<FileStatus> statusIterator; + + /** + * Constructor. + * @param statusIterator an iterator over the remote status entries + */ + LocatedFileStatusIterator(RemoteIterator<FileStatus> statusIterator) { + this.statusIterator = statusIterator; + } + + @Override + public boolean hasNext() throws IOException { + return statusIterator.hasNext(); + } + + @Override + public LocatedFileStatus next() throws IOException { + return owner.toLocatedFileStatus(statusIterator.next()); + } + } + + /** + * Accept all entries except the base path and those which map to S3N + * pseudo directory markers. + */ + static class AcceptAllButSelfAndS3nDirs implements FileStatusAcceptor { + + /** Base path. */ + private final Path qualifiedPath; + + /** + * Constructor. + * @param qualifiedPath an already-qualified path. + */ + public AcceptAllButSelfAndS3nDirs(Path qualifiedPath) { + this.qualifiedPath = qualifiedPath; + } + + /** + * Reject a summary entry if the key path is the qualified Path, or + * it ends with {@code "_$folder$"}. + * @param keyPath key path of the entry + * @param summary summary entry + * @return true if the entry is accepted (i.e. that a status entry + * should be generated.) + */ + @Override + public boolean accept(Path keyPath, S3ObjectSummary summary) { + return !keyPath.equals(qualifiedPath) && + !summary.getKey().endsWith(S3N_FOLDER_SUFFIX); + } + + /** + * Accept all prefixes except the one for the base path, "self". + * @param keyPath qualified path to the entry + * @param prefix common prefix in listing. + * @return true if the entry is accepted (i.e. that a status entry + * should be generated. + */ + @Override + public boolean accept(Path keyPath, String prefix) { + return !keyPath.equals(qualifiedPath); + } + } + + /** + * A Path filter which accepts all filenames. + */ + static final PathFilter ACCEPT_ALL = new PathFilter() { + @Override + public boolean accept(Path file) { + return true; + } + + @Override + public String toString() { + return "ACCEPT_ALL"; + } + }; + +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java index c45f4a5..61727c9 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java @@ -85,6 +85,7 @@ import org.apache.hadoop.util.Progressable; import org.apache.hadoop.util.VersionInfo; import static org.apache.hadoop.fs.s3a.Constants.*; +import static org.apache.hadoop.fs.s3a.Listing.ACCEPT_ALL; import static org.apache.hadoop.fs.s3a.S3AUtils.*; import static org.apache.hadoop.fs.s3a.Statistic.*; @@ -116,6 +117,7 @@ public class S3AFileSystem extends FileSystem { private AmazonS3Client s3; private String bucket; private int maxKeys; + private Listing listing; private long partSize; private boolean enableMultiObjectsDelete; private TransferManager transfers; @@ -184,6 +186,7 @@ public class S3AFileSystem extends FileSystem { initAmazonS3Client(conf, credentials, awsConf); maxKeys = intOption(conf, MAX_PAGING_KEYS, DEFAULT_MAX_PAGING_KEYS, 1); + listing = new Listing(this); partSize = conf.getLong(MULTIPART_SIZE, DEFAULT_MULTIPART_SIZE); if (partSize < 5 * 1024 * 1024) { LOG.error(MULTIPART_SIZE + " must be at least 5 MB"); @@ -524,7 +527,11 @@ public class S3AFileSystem extends FileSystem { super(); } - /* Turns a path (relative or otherwise) into an S3 key + /** + * Turns a path (relative or otherwise) into an S3 key. + * + * @param path input path, may be relative to the working dir + * @return a key excluding the leading "/", or, if it is the root path, "" */ private String pathToKey(Path path) { if (!path.isAbsolute()) { @@ -538,11 +545,50 @@ public class S3AFileSystem extends FileSystem { return path.toUri().getPath().substring(1); } + /** + * Turns a path (relative or otherwise) into an S3 key, adding a trailing + * "/" if the path is not the root <i>and</i> does not already have a "/" + * at the end. + * + * @param key s3 key or "" + * @return the with a trailing "/", or, if it is the root key, "", + */ + private String maybeAddTrailingSlash(String key) { + if (!key.isEmpty() && !key.endsWith("/")) { + return key + '/'; + } else { + return key; + } + } + + /** + * Convert a path back to a key. + * @param key input key + * @return the path from this key + */ private Path keyToPath(String key) { return new Path("/" + key); } /** + * Convert a key to a fully qualified path. + * @param key input key + * @return the fully qualified path including URI scheme and bucket name. + */ + Path keyToQualifiedPath(String key) { + return qualify(keyToPath(key)); + } + + /** + * Qualify a path. + * @param path path to qualify + * @return a qualified path. + */ + Path qualify(Path path) { + return path.makeQualified(uri, workingDir); + } + + /** * Check that a Path belongs to this FileSystem. * Unlike the superclass, this version does not look at authority, * only hostnames. @@ -560,10 +606,10 @@ public class S3AFileSystem extends FileSystem { } /** - * Opens an FSDataInputStream at the indicated Path. - * @param f the file name to open - * @param bufferSize the size of the buffer to be used. - */ + * Opens an FSDataInputStream at the indicated Path. + * @param f the file name to open + * @param bufferSize the size of the buffer to be used. + */ public FSDataInputStream open(Path f, int bufferSize) throws IOException { @@ -885,7 +931,7 @@ public class S3AFileSystem extends FileSystem { * @return the next result object */ protected ObjectListing continueListObjects(ObjectListing objects) { - incrementStatistic(OBJECT_LIST_REQUESTS); + incrementStatistic(OBJECT_CONTINUE_LIST_REQUESTS); incrementReadOperations(); return s3.listNextBatchOfObjects(objects); } @@ -1142,12 +1188,7 @@ public class S3AFileSystem extends FileSystem { } else { LOG.debug("Getting objects for directory prefix {} to delete", key); - ListObjectsRequest request = new ListObjectsRequest(); - request.setBucketName(bucket); - request.setPrefix(key); - // Hopefully not setting a delimiter will cause this to find everything - //request.setDelimiter("/"); - request.setMaxKeys(maxKeys); + ListObjectsRequest request = createListObjectsRequest(key, null); ObjectListing objects = listObjects(request); List<DeleteObjectsRequest.KeyVersion> keys = @@ -1221,66 +1262,57 @@ public class S3AFileSystem extends FileSystem { */ public FileStatus[] innerListStatus(Path f) throws FileNotFoundException, IOException, AmazonClientException { - String key = pathToKey(f); - LOG.debug("List status for path: {}", f); + Path path = qualify(f); + String key = pathToKey(path); + LOG.debug("List status for path: {}", path); incrementStatistic(INVOCATION_LIST_STATUS); - final List<FileStatus> result = new ArrayList<FileStatus>(); - final FileStatus fileStatus = getFileStatus(f); + List<FileStatus> result; + final FileStatus fileStatus = getFileStatus(path); if (fileStatus.isDirectory()) { if (!key.isEmpty()) { - key = key + "/"; + key = key + '/'; } - ListObjectsRequest request = new ListObjectsRequest(); - request.setBucketName(bucket); - request.setPrefix(key); - request.setDelimiter("/"); - request.setMaxKeys(maxKeys); - + ListObjectsRequest request = createListObjectsRequest(key, "/"); LOG.debug("listStatus: doing listObjects for directory {}", key); - ObjectListing objects = listObjects(request); - - Path fQualified = f.makeQualified(uri, workingDir); - - while (true) { - for (S3ObjectSummary summary : objects.getObjectSummaries()) { - Path keyPath = keyToPath(summary.getKey()).makeQualified(uri, workingDir); - // Skip over keys that are ourselves and old S3N _$folder$ files - if (keyPath.equals(fQualified) || - summary.getKey().endsWith(S3N_FOLDER_SUFFIX)) { - LOG.debug("Ignoring: {}", keyPath); - } else { - S3AFileStatus status = createFileStatus(keyPath, summary, - getDefaultBlockSize(keyPath)); - result.add(status); - LOG.debug("Adding: {}", status); - } - } - - for (String prefix : objects.getCommonPrefixes()) { - Path keyPath = keyToPath(prefix).makeQualified(uri, workingDir); - if (!keyPath.equals(f)) { - result.add(new S3AFileStatus(true, false, keyPath)); - LOG.debug("Adding: rd: {}", keyPath); - } - } - - if (objects.isTruncated()) { - LOG.debug("listStatus: list truncated - getting next batch"); - objects = continueListObjects(objects); - } else { - break; - } + Listing.FileStatusListingIterator files = + listing.createFileStatusListingIterator(path, + request, + ACCEPT_ALL, + new Listing.AcceptAllButSelfAndS3nDirs(path)); + result = new ArrayList<>(files.getBatchSize()); + while (files.hasNext()) { + result.add(files.next()); } + return result.toArray(new FileStatus[result.size()]); } else { - LOG.debug("Adding: rd (not a dir): {}", f); - result.add(fileStatus); + LOG.debug("Adding: rd (not a dir): {}", path); + FileStatus[] stats = new FileStatus[1]; + stats[0]= fileStatus; + return stats; } + } - return result.toArray(new FileStatus[result.size()]); + /** + * Create a {@code ListObjectsRequest} request against this bucket, + * with the maximum keys returned in a query set by {@link #maxKeys}. + * @param key key for request + * @param delimiter any delimiter + * @return the request + */ + private ListObjectsRequest createListObjectsRequest(String key, + String delimiter) { + ListObjectsRequest request = new ListObjectsRequest(); + request.setBucketName(bucket); + request.setMaxKeys(maxKeys); + request.setPrefix(key); + if (delimiter != null) { + request.setDelimiter(delimiter); + } + return request; } /** @@ -1381,11 +1413,11 @@ public class S3AFileSystem extends FileSystem { * @throws java.io.FileNotFoundException when the path does not exist; * @throws IOException on other problems. */ - public S3AFileStatus getFileStatus(Path f) throws IOException { - String key = pathToKey(f); + public S3AFileStatus getFileStatus(final Path f) throws IOException { incrementStatistic(INVOCATION_GET_FILE_STATUS); - LOG.debug("Getting path status for {} ({})", f , key); - + final Path path = qualify(f); + String key = pathToKey(path); + LOG.debug("Getting path status for {} ({})", path , key); if (!key.isEmpty()) { try { ObjectMetadata meta = getObjectMetadata(key); @@ -1393,20 +1425,20 @@ public class S3AFileSystem extends FileSystem { if (objectRepresentsDirectory(key, meta.getContentLength())) { LOG.debug("Found exact file: fake directory"); return new S3AFileStatus(true, true, - f.makeQualified(uri, workingDir)); + path); } else { LOG.debug("Found exact file: normal file"); return new S3AFileStatus(meta.getContentLength(), dateToLong(meta.getLastModified()), - f.makeQualified(uri, workingDir), - getDefaultBlockSize(f.makeQualified(uri, workingDir))); + path, + getDefaultBlockSize(path)); } } catch (AmazonServiceException e) { if (e.getStatusCode() != 404) { - throw translateException("getFileStatus", f, e); + throw translateException("getFileStatus", path, e); } } catch (AmazonClientException e) { - throw translateException("getFileStatus", f, e); + throw translateException("getFileStatus", path, e); } // Necessary? @@ -1417,14 +1449,14 @@ public class S3AFileSystem extends FileSystem { if (objectRepresentsDirectory(newKey, meta.getContentLength())) { LOG.debug("Found file (with /): fake directory"); - return new S3AFileStatus(true, true, f.makeQualified(uri, workingDir)); + return new S3AFileStatus(true, true, path); } else { LOG.warn("Found file (with /): real file? should not happen: {}", key); return new S3AFileStatus(meta.getContentLength(), dateToLong(meta.getLastModified()), - f.makeQualified(uri, workingDir), - getDefaultBlockSize(f.makeQualified(uri, workingDir))); + path, + getDefaultBlockSize(path)); } } catch (AmazonServiceException e) { if (e.getStatusCode() != 404) { @@ -1437,9 +1469,7 @@ public class S3AFileSystem extends FileSystem { } try { - if (!key.isEmpty() && !key.endsWith("/")) { - key = key + "/"; - } + key = maybeAddTrailingSlash(key); ListObjectsRequest request = new ListObjectsRequest(); request.setBucketName(bucket); request.setPrefix(key); @@ -1463,11 +1493,10 @@ public class S3AFileSystem extends FileSystem { } } - return new S3AFileStatus(true, false, - f.makeQualified(uri, workingDir)); + return new S3AFileStatus(true, false, path); } else if (key.isEmpty()) { LOG.debug("Found root directory"); - return new S3AFileStatus(true, true, f.makeQualified(uri, workingDir)); + return new S3AFileStatus(true, true, path); } } catch (AmazonServiceException e) { if (e.getStatusCode() != 404) { @@ -1477,8 +1506,8 @@ public class S3AFileSystem extends FileSystem { throw translateException("getFileStatus", key, e); } - LOG.debug("Not Found: {}", f); - throw new FileNotFoundException("No such file or directory: " + f); + LOG.debug("Not Found: {}", path); + throw new FileNotFoundException("No such file or directory: " + path); } /** @@ -1839,7 +1868,15 @@ public class S3AFileSystem extends FileSystem { } /** - * Override superclass so as to add statistic collection. + * Get the maximum key count. + * @return a value, valid after initialization + */ + int getMaxKeys() { + return maxKeys; + } + + /** + * Increments the statistic {@link Statistic#INVOCATION_GLOB_STATUS}. * {@inheritDoc} */ @Override @@ -1864,24 +1901,6 @@ public class S3AFileSystem extends FileSystem { * {@inheritDoc} */ @Override - public RemoteIterator<LocatedFileStatus> listLocatedStatus(Path f) - throws FileNotFoundException, IOException { - incrementStatistic(INVOCATION_LIST_LOCATED_STATUS); - return super.listLocatedStatus(f); - } - - @Override - public RemoteIterator<LocatedFileStatus> listFiles(Path f, - boolean recursive) throws FileNotFoundException, IOException { - incrementStatistic(INVOCATION_LIST_FILES); - return super.listFiles(f, recursive); - } - - /** - * Override superclass so as to add statistic collection. - * {@inheritDoc} - */ - @Override public boolean exists(Path f) throws IOException { incrementStatistic(INVOCATION_EXISTS); return super.exists(f); @@ -1908,6 +1927,129 @@ public class S3AFileSystem extends FileSystem { } /** + * {@inheritDoc}. + * + * This implementation is optimized for S3, which can do a bulk listing + * off all entries under a path in one single operation. Thus there is + * no need to recursively walk the directory tree. + * + * Instead a {@link ListObjectsRequest} is created requesting a (windowed) + * listing of all entries under the given path. This is used to construct + * an {@code ObjectListingIterator} instance, iteratively returning the + * sequence of lists of elements under the path. This is then iterated + * over in a {@code FileStatusListingIterator}, which generates + * {@link S3AFileStatus} instances, one per listing entry. + * These are then translated into {@link LocatedFileStatus} instances. + * + * This is essentially a nested and wrapped set of iterators, with some + * generator classes; an architecture which may become less convoluted + * using lambda-expressions. + * @param f a path + * @param recursive if the subdirectories need to be traversed recursively + * + * @return an iterator that traverses statuses of the files/directories + * in the given path + * @throws FileNotFoundException if {@code path} does not exist + * @throws IOException if any I/O error occurred + */ + @Override + public RemoteIterator<LocatedFileStatus> listFiles(Path f, + boolean recursive) throws FileNotFoundException, IOException { + incrementStatistic(INVOCATION_LIST_FILES); + Path path = qualify(f); + LOG.debug("listFiles({}, {})", path, recursive); + try { + // lookup dir triggers existence check + final FileStatus fileStatus = getFileStatus(path); + if (fileStatus.isFile()) { + // simple case: File + LOG.debug("Path is a file"); + return new Listing.SingleStatusRemoteIterator( + toLocatedFileStatus(fileStatus)); + } else { + // directory: do a bulk operation + String key = maybeAddTrailingSlash(pathToKey(path)); + String delimiter = recursive ? null : "/"; + LOG.debug("Requesting all entries under {} with delimiter '{}'", + key, delimiter); + return listing.createLocatedFileStatusIterator( + listing.createFileStatusListingIterator(path, + createListObjectsRequest(key, delimiter), + ACCEPT_ALL, + new Listing.AcceptFilesOnly(path))); + } + } catch (AmazonClientException e) { + throw translateException("listFiles", path, e); + } + } + + /** + * Override superclass so as to add statistic collection. + * {@inheritDoc} + */ + @Override + public RemoteIterator<LocatedFileStatus> listLocatedStatus(Path f) + throws FileNotFoundException, IOException { + return listLocatedStatus(f, ACCEPT_ALL); + } + + /** + * {@inheritDoc}. + * + * S3 Optimized directory listing. The initial operation performs the + * first bulk listing; extra listings will take place + * when all the current set of results are used up. + * @param f a path + * @param filter a path filter + * @return an iterator that traverses statuses of the files/directories + * in the given path + * @throws FileNotFoundException if {@code path} does not exist + * @throws IOException if any I/O error occurred + */ + @Override + public RemoteIterator<LocatedFileStatus> listLocatedStatus(final Path f, + final PathFilter filter) + throws FileNotFoundException, IOException { + incrementStatistic(INVOCATION_LIST_LOCATED_STATUS); + Path path = qualify(f); + LOG.debug("listLocatedStatus({}, {}", path, filter); + try { + // lookup dir triggers existence check + final FileStatus fileStatus = getFileStatus(path); + if (fileStatus.isFile()) { + // simple case: File + LOG.debug("Path is a file"); + return new Listing.SingleStatusRemoteIterator( + filter.accept(path) ? toLocatedFileStatus(fileStatus) : null); + } else { + // directory: trigger a lookup + String key = maybeAddTrailingSlash(pathToKey(path)); + return listing.createLocatedFileStatusIterator( + listing.createFileStatusListingIterator(path, + createListObjectsRequest(key, "/"), + filter, + new Listing.AcceptAllButSelfAndS3nDirs(path))); + } + } catch (AmazonClientException e) { + throw translateException("listLocatedStatus", path, e); + } + } + + /** + * Build a {@link LocatedFileStatus} from a {@link FileStatus} instance. + * @param status file status + * @return a located status with block locations set up from this FS. + * @throws IOException IO Problems. + */ + LocatedFileStatus toLocatedFileStatus(FileStatus status) + throws IOException { + return new LocatedFileStatus(status, + status.isFile() ? + getFileBlockLocations(status, 0, status.getLen()) + : null); + } + + /** * Get a integer option >= the minimum allowed value. * @param conf configuration * @param key key to look up @@ -1944,38 +2086,4 @@ public class S3AFileSystem extends FileSystem { return v; } - /** - * This is a simple encapsulation of the - * S3 access key and secret. - */ - static class AWSAccessKeys { - private String accessKey = null; - private String accessSecret = null; - - /** - * Constructor. - * @param key - AWS access key - * @param secret - AWS secret key - */ - public AWSAccessKeys(String key, String secret) { - accessKey = key; - accessSecret = secret; - } - - /** - * Return the AWS access key. - * @return key - */ - public String getAccessKey() { - return accessKey; - } - - /** - * Return the AWS secret key. - * @return secret - */ - public String getAccessSecret() { - return accessSecret; - } - } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java index 514b974..b4c4063 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AInstrumentation.java @@ -95,6 +95,7 @@ public class S3AInstrumentation { OBJECT_COPY_REQUESTS, OBJECT_DELETE_REQUESTS, OBJECT_LIST_REQUESTS, + OBJECT_CONTINUE_LIST_REQUESTS, OBJECT_METADATA_REQUESTS, OBJECT_MULTIPART_UPLOAD_ABORTED, OBJECT_PUT_BYTES, http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java index 8033ac3..3d48fcd 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AUtils.java @@ -280,4 +280,16 @@ public final class S3AUtils { return val; } } + + /** + * String information about a summary entry for debug messages. + * @param summary summary object + * @return string value + */ + public static String stringify(S3ObjectSummary summary) { + StringBuilder builder = new StringBuilder(summary.getKey().length() + 100); + builder.append(summary.getKey()).append(' '); + builder.append("size=").append(summary.getSize()); + return builder.toString(); + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java index 36d163c..cbc34d6 100644 --- a/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java +++ b/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Statistic.java @@ -69,6 +69,8 @@ public enum Statistic { OBJECT_DELETE_REQUESTS("object_delete_requests", "Object delete requests"), OBJECT_LIST_REQUESTS("object_list_requests", "Number of object listings made"), + OBJECT_CONTINUE_LIST_REQUESTS("object_continue_list_requests", + "Number of continued object listings made"), OBJECT_METADATA_REQUESTS("object_metadata_requests", "Number of requests for object metadata"), OBJECT_MULTIPART_UPLOAD_ABORTED("object_multipart_aborted", http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java index d7b8fe3..5937d49 100644 --- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java +++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/contract/s3a/TestS3AContractGetFileStatus.java @@ -20,6 +20,8 @@ package org.apache.hadoop.fs.contract.s3a; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.contract.AbstractFSContract; import org.apache.hadoop.fs.contract.AbstractContractGetFileStatusTest; +import org.apache.hadoop.fs.s3a.Constants; +import org.apache.hadoop.fs.s3a.S3ATestUtils; public class TestS3AContractGetFileStatus extends AbstractContractGetFileStatusTest { @@ -28,4 +30,18 @@ public class TestS3AContractGetFileStatus extends AbstractContractGetFileStatusT return new S3AContract(conf); } + @Override + public void teardown() throws Exception { + getLog().info("FS details {}", getFileSystem()); + super.teardown(); + } + + @Override + protected Configuration createConfiguration() { + Configuration conf = super.createConfiguration(); + S3ATestUtils.disableFilesystemCaching(conf); + // aggressively low page size forces tests to go multipage + conf.setInt(Constants.MAX_PAGING_KEYS, 2); + return conf; + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java index 39c6028..e45db48 100644 --- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java +++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/S3ATestUtils.java @@ -124,7 +124,7 @@ public class S3ATestUtils { try { callback.call(); return; - } catch (FailFastException e) { + } catch (InterruptedException | FailFastException e) { throw e; } catch (Exception e) { lastException = e; @@ -321,7 +321,7 @@ public class S3ATestUtils { * @return true if the value is {@code ==} the other's */ public boolean diffEquals(MetricDiff that) { - return this.currentValue() == that.currentValue(); + return this.diff() == that.diff(); } /** @@ -330,7 +330,7 @@ public class S3ATestUtils { * @return true if the value is {@code <} the other's */ public boolean diffLessThan(MetricDiff that) { - return this.currentValue() < that.currentValue(); + return this.diff() < that.diff(); } /** @@ -339,7 +339,7 @@ public class S3ATestUtils { * @return true if the value is {@code <=} the other's */ public boolean diffLessThanOrEquals(MetricDiff that) { - return this.currentValue() <= that.currentValue(); + return this.diff() <= that.diff(); } /** http://git-wip-us.apache.org/repos/asf/hadoop/blob/986162f9/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java ---------------------------------------------------------------------- diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java index 7ece394..35ea3ad 100644 --- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java +++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/scale/TestS3ADirectoryPerformance.java @@ -18,7 +18,6 @@ package org.apache.hadoop.fs.s3a.scale; -import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.s3a.Statistic; import org.junit.Test; @@ -51,6 +50,8 @@ public class TestS3ADirectoryPerformance extends S3AScaleTestBase { int files = scale; MetricDiff metadataRequests = new MetricDiff(fs, OBJECT_METADATA_REQUESTS); MetricDiff listRequests = new MetricDiff(fs, OBJECT_LIST_REQUESTS); + MetricDiff listContinueRequests = + new MetricDiff(fs, OBJECT_CONTINUE_LIST_REQUESTS); MetricDiff listStatusCalls = new MetricDiff(fs, INVOCATION_LIST_FILES); MetricDiff getFileStatusCalls = new MetricDiff(fs, INVOCATION_GET_FILE_STATUS); @@ -69,34 +70,29 @@ public class TestS3ADirectoryPerformance extends S3AScaleTestBase { printThenReset(LOG, metadataRequests, listRequests, + listContinueRequests, listStatusCalls, getFileStatusCalls); + describe("Listing files via treewalk"); try { // Scan the directory via an explicit tree walk. // This is the baseline for any listing speedups. - MetricDiff treewalkMetadataRequests = - new MetricDiff(fs, OBJECT_METADATA_REQUESTS); - MetricDiff treewalkListRequests = new MetricDiff(fs, - OBJECT_LIST_REQUESTS); - MetricDiff treewalkListStatusCalls = new MetricDiff(fs, - INVOCATION_LIST_FILES); - MetricDiff treewalkGetFileStatusCalls = - new MetricDiff(fs, INVOCATION_GET_FILE_STATUS); NanoTimer treeWalkTimer = new NanoTimer(); TreeScanResults treewalkResults = treeWalk(fs, listDir); - treeWalkTimer.end("List status via treewalk"); + treeWalkTimer.end("List status via treewalk of %s", created); - print(LOG, - treewalkMetadataRequests, - treewalkListRequests, - treewalkListStatusCalls, - treewalkGetFileStatusCalls); + printThenReset(LOG, + metadataRequests, + listRequests, + listContinueRequests, + listStatusCalls, + getFileStatusCalls); assertEquals("Files found in listFiles(recursive=true) " + " created=" + created + " listed=" + treewalkResults, created.getFileCount(), treewalkResults.getFileCount()); - + describe("Listing files via listFiles(recursive=true)"); // listFiles() does the recursion internally NanoTimer listFilesRecursiveTimer = new NanoTimer(); @@ -108,31 +104,33 @@ public class TestS3ADirectoryPerformance extends S3AScaleTestBase { " created=" + created + " listed=" + listFilesResults, created.getFileCount(), listFilesResults.getFileCount()); - treewalkListRequests.assertDiffEquals(listRequests); - printThenReset(LOG, - metadataRequests, listRequests, - listStatusCalls, getFileStatusCalls); - - NanoTimer globStatusTimer = new NanoTimer(); - FileStatus[] globStatusFiles = fs.globStatus(listDir); - globStatusTimer.end("Time to globStatus() %s", globStatusTimer); - LOG.info("Time for glob status {} entries: {}", - globStatusFiles.length, - toHuman(createTimer.duration())); - printThenReset(LOG, + // only two list operations should have taken place + print(LOG, metadataRequests, listRequests, + listContinueRequests, listStatusCalls, getFileStatusCalls); + assertEquals(listRequests.toString(), 2, listRequests.diff()); + reset(metadataRequests, + listRequests, + listContinueRequests, + listStatusCalls, + getFileStatusCalls); + } finally { + describe("deletion"); // deletion at the end of the run NanoTimer deleteTimer = new NanoTimer(); fs.delete(listDir, true); deleteTimer.end("Deleting directory tree"); printThenReset(LOG, - metadataRequests, listRequests, - listStatusCalls, getFileStatusCalls); + metadataRequests, + listRequests, + listContinueRequests, + listStatusCalls, + getFileStatusCalls); } } --------------------------------------------------------------------- To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-commits-h...@hadoop.apache.org
