HDFS-12222. Document and test BlockLocation for erasure-coded files. Contributed by Huafeng Wang.

This commit is contained in:
Andrew Wang 2017-09-12 17:35:30 -07:00
parent a1c398e0dc
commit f4b6267465
11 changed files with 333 additions and 5 deletions

View File

@ -928,6 +928,11 @@ public abstract class AbstractFileSystem {
* The specification of this method matches that of * The specification of this method matches that of
* {@link FileContext#listLocatedStatus(Path)} except that Path f * {@link FileContext#listLocatedStatus(Path)} except that Path f
* must be for this file system. * must be for this file system.
*
* In HDFS implementation, the BlockLocation of returned LocatedFileStatus
* will have different formats for replicated and erasure coded file. Please
* refer to {@link FileSystem#getFileBlockLocations(FileStatus, long, long)}
* for more details.
*/ */
public RemoteIterator<LocatedFileStatus> listLocatedStatus(final Path f) public RemoteIterator<LocatedFileStatus> listLocatedStatus(final Path f)
throws AccessControlException, FileNotFoundException, throws AccessControlException, FileNotFoundException,

View File

@ -28,6 +28,34 @@ import org.apache.hadoop.util.StringInterner;
* Represents the network location of a block, information about the hosts * Represents the network location of a block, information about the hosts
* that contain block replicas, and other block metadata (E.g. the file * that contain block replicas, and other block metadata (E.g. the file
* offset associated with the block, length, whether it is corrupt, etc). * offset associated with the block, length, whether it is corrupt, etc).
*
* For a single BlockLocation, it will have different meanings for replicated
* and erasure coded files.
*
* If the file is 3-replicated, offset and length of a BlockLocation represent
* the absolute value in the file and the hosts are the 3 datanodes that
* holding the replicas. Here is an example:
* <pre>
* BlockLocation(offset: 0, length: BLOCK_SIZE,
* hosts: {"host1:9866", "host2:9866, host3:9866"})
* </pre>
*
* And if the file is erasure-coded, each BlockLocation represents a logical
* block groups. Value offset is the offset of a block group in the file and
* value length is the total length of a block group. Hosts of a BlockLocation
* are the datanodes that holding all the data blocks and parity blocks of a
* block group.
* Suppose we have a RS_3_2 coded file (3 data units and 2 parity units).
* A BlockLocation example will be like:
* <pre>
* BlockLocation(offset: 0, length: 3 * BLOCK_SIZE, hosts: {"host1:9866",
* "host2:9866","host3:9866","host4:9866","host5:9866"})
* </pre>
*
* Please refer to
* {@link FileSystem#getFileBlockLocations(FileStatus, long, long)} or
* {@link FileContext#getFileBlockLocations(Path, long, long)}
* for more examples.
*/ */
@InterfaceAudience.Public @InterfaceAudience.Public
@InterfaceStability.Stable @InterfaceStability.Stable

View File

@ -1294,6 +1294,35 @@ public class FileContext {
* This call is most helpful with DFS, where it returns * This call is most helpful with DFS, where it returns
* hostnames of machines that contain the given file. * hostnames of machines that contain the given file.
* *
* In HDFS, if file is three-replicated, the returned array contains
* elements like:
* <pre>
* BlockLocation(offset: 0, length: BLOCK_SIZE,
* hosts: {"host1:9866", "host2:9866, host3:9866"})
* BlockLocation(offset: BLOCK_SIZE, length: BLOCK_SIZE,
* hosts: {"host2:9866", "host3:9866, host4:9866"})
* </pre>
*
* And if a file is erasure-coded, the returned BlockLocation are logical
* block groups.
*
* Suppose we have a RS_3_2 coded file (3 data units and 2 parity units).
* 1. If the file size is less than one stripe size, say 2 * CELL_SIZE, then
* there will be one BlockLocation returned, with 0 offset, actual file size
* and 4 hosts (2 data blocks and 2 parity blocks) hosting the actual blocks.
* 3. If the file size is less than one group size but greater than one
* stripe size, then there will be one BlockLocation returned, with 0 offset,
* actual file size with 5 hosts (3 data blocks and 2 parity blocks) hosting
* the actual blocks.
* 4. If the file size is greater than one group size, 3 * BLOCK_SIZE + 123
* for example, then the result will be like:
* <pre>
* BlockLocation(offset: 0, length: 3 * BLOCK_SIZE, hosts: {"host1:9866",
* "host2:9866","host3:9866","host4:9866","host5:9866"})
* BlockLocation(offset: 3 * BLOCK_SIZE, length: 123, hosts: {"host1:9866",
* "host4:9866", "host5:9866"})
* </pre>
*
* @param f - get blocklocations of this file * @param f - get blocklocations of this file
* @param start position (byte offset) * @param start position (byte offset)
* @param len (in bytes) * @param len (in bytes)

View File

@ -799,7 +799,36 @@ public abstract class FileSystem extends Configured implements Closeable {
* The default implementation returns an array containing one element: * The default implementation returns an array containing one element:
* <pre> * <pre>
* BlockLocation( { "localhost:9866" }, { "localhost" }, 0, file.getLen()) * BlockLocation( { "localhost:9866" }, { "localhost" }, 0, file.getLen())
* </pre>> * </pre>
*
* In HDFS, if file is three-replicated, the returned array contains
* elements like:
* <pre>
* BlockLocation(offset: 0, length: BLOCK_SIZE,
* hosts: {"host1:9866", "host2:9866, host3:9866"})
* BlockLocation(offset: BLOCK_SIZE, length: BLOCK_SIZE,
* hosts: {"host2:9866", "host3:9866, host4:9866"})
* </pre>
*
* And if a file is erasure-coded, the returned BlockLocation are logical
* block groups.
*
* Suppose we have a RS_3_2 coded file (3 data units and 2 parity units).
* 1. If the file size is less than one stripe size, say 2 * CELL_SIZE, then
* there will be one BlockLocation returned, with 0 offset, actual file size
* and 4 hosts (2 data blocks and 2 parity blocks) hosting the actual blocks.
* 3. If the file size is less than one group size but greater than one
* stripe size, then there will be one BlockLocation returned, with 0 offset,
* actual file size with 5 hosts (3 data blocks and 2 parity blocks) hosting
* the actual blocks.
* 4. If the file size is greater than one group size, 3 * BLOCK_SIZE + 123
* for example, then the result will be like:
* <pre>
* BlockLocation(offset: 0, length: 3 * BLOCK_SIZE, hosts: {"host1:9866",
* "host2:9866","host3:9866","host4:9866","host5:9866"})
* BlockLocation(offset: 3 * BLOCK_SIZE, length: 123, hosts: {"host1:9866",
* "host4:9866", "host5:9866"})
* </pre>
* *
* @param file FilesStatus to get data from * @param file FilesStatus to get data from
* @param start offset into the given file * @param start offset into the given file
@ -2115,6 +2144,7 @@ public abstract class FileSystem extends Configured implements Closeable {
* List the statuses and block locations of the files in the given path. * List the statuses and block locations of the files in the given path.
* Does not guarantee to return the iterator that traverses statuses * Does not guarantee to return the iterator that traverses statuses
* of the files in a sorted order. * of the files in a sorted order.
*
* <pre> * <pre>
* If the path is a directory, * If the path is a directory,
* if recursive is false, returns files in the directory; * if recursive is false, returns files in the directory;

View File

@ -123,6 +123,13 @@ public class LocatedFileStatus extends FileStatus {
/** /**
* Get the file's block locations * Get the file's block locations
*
* In HDFS, the returned BlockLocation will have different formats for
* replicated and erasure coded file.
* Please refer to
* {@link FileSystem#getFileBlockLocations(FileStatus, long, long)}
* for more details.
*
* @return the file's block locations * @return the file's block locations
*/ */
public BlockLocation[] getBlockLocations() { public BlockLocation[] getBlockLocations() {

View File

@ -115,6 +115,14 @@ public class Hdfs extends AbstractFileSystem {
return dfs.delete(getUriPath(f), recursive); return dfs.delete(getUriPath(f), recursive);
} }
/**
* The returned BlockLocation will have different formats for replicated
* and erasure coded file.
*
* Please refer to
* {@link FileContext#getFileBlockLocations(Path, long, long)}
* for more details.
*/
@Override @Override
public BlockLocation[] getFileBlockLocations(Path p, long start, long len) public BlockLocation[] getFileBlockLocations(Path p, long start, long len)
throws IOException, UnresolvedLinkException { throws IOException, UnresolvedLinkException {
@ -165,6 +173,13 @@ public class Hdfs extends AbstractFileSystem {
return dfs.getServerDefaults(); return dfs.getServerDefaults();
} }
/**
* The BlockLocation of returned LocatedFileStatus will have different
* formats for replicated and erasure coded file.
* Please refer to
* {@link FileContext#getFileBlockLocations(Path, long, long)} for
* more details.
*/
@Override @Override
public RemoteIterator<LocatedFileStatus> listLocatedStatus( public RemoteIterator<LocatedFileStatus> listLocatedStatus(
final Path p) final Path p)

View File

@ -72,6 +72,7 @@ import org.apache.hadoop.fs.ContentSummary;
import org.apache.hadoop.fs.CreateFlag; import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.FileAlreadyExistsException; import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.FileEncryptionInfo; import org.apache.hadoop.fs.FileEncryptionInfo;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsServerDefaults; import org.apache.hadoop.fs.FsServerDefaults;
import org.apache.hadoop.fs.FsStatus; import org.apache.hadoop.fs.FsStatus;
@ -866,6 +867,10 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
* data-placement when performing operations. For example, the * data-placement when performing operations. For example, the
* MapReduce system tries to schedule tasks on the same machines * MapReduce system tries to schedule tasks on the same machines
* as the data-block the task processes. * as the data-block the task processes.
*
* Please refer to
* {@link FileSystem#getFileBlockLocations(FileStatus, long, long)}
* for more details.
*/ */
public BlockLocation[] getBlockLocations(String src, long start, public BlockLocation[] getBlockLocations(String src, long start,
long length) throws IOException { long length) throws IOException {

View File

@ -240,6 +240,13 @@ public class DistributedFileSystem extends FileSystem {
return getFileBlockLocations(file.getPath(), start, len); return getFileBlockLocations(file.getPath(), start, len);
} }
/**
* The returned BlockLocation will have different formats for replicated
* and erasure coded file.
* Please refer to
* {@link FileSystem#getFileBlockLocations(FileStatus, long, long)}
* for more details.
*/
@Override @Override
public BlockLocation[] getFileBlockLocations(Path p, public BlockLocation[] getFileBlockLocations(Path p,
final long start, final long len) throws IOException { final long start, final long len) throws IOException {
@ -1040,6 +1047,13 @@ public class DistributedFileSystem extends FileSystem {
}.resolve(this, absF); }.resolve(this, absF);
} }
/**
* The BlockLocation of returned LocatedFileStatus will have different
* formats for replicated and erasure coded file.
* Please refer to
* {@link FileSystem#getFileBlockLocations(FileStatus, long, long)} for
* more details.
*/
@Override @Override
protected RemoteIterator<LocatedFileStatus> listLocatedStatus(final Path p, protected RemoteIterator<LocatedFileStatus> listLocatedStatus(final Path p,
final PathFilter filter) final PathFilter filter)

View File

@ -78,6 +78,17 @@ public class HdfsLocatedFileStatus extends HdfsFileStatus {
return locations; return locations;
} }
/**
* This function is used to transform the underlying HDFS LocatedBlocks to
* BlockLocations.
*
* The returned BlockLocation will have different formats for replicated
* and erasure coded file.
* Please refer to
* {@link org.apache.hadoop.fs.FileSystem#getFileBlockLocations
* (FileStatus, long, long)}
* for examples.
*/
public final LocatedFileStatus makeQualifiedLocated(URI defaultUri, public final LocatedFileStatus makeQualifiedLocated(URI defaultUri,
Path path) { Path path) {
makeQualified(defaultUri, path); makeQualified(defaultUri, path);
@ -96,5 +107,4 @@ public class HdfsLocatedFileStatus extends HdfsFileStatus {
// satisfy findbugs // satisfy findbugs
return super.hashCode(); return super.hashCode();
} }
} }

View File

@ -50,7 +50,6 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation; import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.CommonConfigurationKeys; import org.apache.hadoop.fs.CommonConfigurationKeys;

View File

@ -0,0 +1,186 @@
/**
* 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.hdfs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import static org.junit.Assert.assertTrue;
/**
* Testing correctness of FileSystem.getFileBlockLocations and
* FileSystem.listFiles for erasure coded files.
*/
public class TestDistributedFileSystemWithECFile {
private final ErasureCodingPolicy ecPolicy =
StripedFileTestUtil.getDefaultECPolicy();
private final int cellSize = ecPolicy.getCellSize();
private final short dataBlocks = (short) ecPolicy.getNumDataUnits();
private final short parityBlocks = (short) ecPolicy.getNumParityUnits();
private final int numDNs = dataBlocks + parityBlocks;
private final int stripesPerBlock = 4;
private final int blockSize = stripesPerBlock * cellSize;
private final int blockGroupSize = blockSize * dataBlocks;
private MiniDFSCluster cluster;
private FileContext fileContext;
private DistributedFileSystem fs;
private Configuration conf = new HdfsConfiguration();
@Before
public void setup() throws IOException {
conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize);
conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_REDUNDANCY_CONSIDERLOAD_KEY,
false);
conf.set(DFSConfigKeys.DFS_NAMENODE_EC_POLICIES_ENABLED_KEY,
StripedFileTestUtil.getDefaultECPolicy().getName());
cluster = new MiniDFSCluster.Builder(conf).numDataNodes(numDNs).build();
fileContext = FileContext.getFileContext(cluster.getURI(0), conf);
fs = cluster.getFileSystem();
fs.mkdirs(new Path("/ec"));
cluster.getFileSystem().getClient().setErasureCodingPolicy("/ec",
StripedFileTestUtil.getDefaultECPolicy().getName());
}
@After
public void tearDown() throws IOException {
if (cluster != null) {
cluster.shutdown();
cluster = null;
}
}
private void createFile(String path, int size) throws Exception {
byte[] expected = StripedFileTestUtil.generateBytes(size);
Path src = new Path(path);
DFSTestUtil.writeFile(fs, src, new String(expected));
StripedFileTestUtil.waitBlockGroupsReported(fs, src.toString());
StripedFileTestUtil.verifyLength(fs, src, size);
}
@Test(timeout=60000)
public void testListECFilesSmallerThanOneCell() throws Exception {
createFile("/ec/smallcell", 1);
final List<LocatedFileStatus> retVal = new ArrayList<>();
final RemoteIterator<LocatedFileStatus> iter =
cluster.getFileSystem().listFiles(new Path("/ec"), true);
while (iter.hasNext()) {
retVal.add(iter.next());
}
assertTrue(retVal.size() == 1);
LocatedFileStatus fileStatus = retVal.get(0);
assertSmallerThanOneCell(fileStatus.getBlockLocations());
BlockLocation[] locations = cluster.getFileSystem().getFileBlockLocations(
fileStatus, 0, fileStatus.getLen());
assertSmallerThanOneCell(locations);
//Test FileContext
fileStatus = fileContext.listLocatedStatus(new Path("/ec")).next();
assertSmallerThanOneCell(fileStatus.getBlockLocations());
locations = fileContext.getFileBlockLocations(new Path("/ec/smallcell"),
0, fileStatus.getLen());
assertSmallerThanOneCell(locations);
}
private void assertSmallerThanOneCell(BlockLocation[] locations)
throws IOException {
assertTrue(locations.length == 1);
BlockLocation blockLocation = locations[0];
assertTrue(blockLocation.getOffset() == 0);
assertTrue(blockLocation.getLength() == 1);
assertTrue(blockLocation.getHosts().length == 1 + parityBlocks);
}
@Test(timeout=60000)
public void testListECFilesSmallerThanOneStripe() throws Exception {
int dataBlocksNum = 3;
createFile("/ec/smallstripe", cellSize * dataBlocksNum);
RemoteIterator<LocatedFileStatus> iter =
cluster.getFileSystem().listFiles(new Path("/ec"), true);
LocatedFileStatus fileStatus = iter.next();
assertSmallerThanOneStripe(fileStatus.getBlockLocations(), dataBlocksNum);
BlockLocation[] locations = cluster.getFileSystem().getFileBlockLocations(
fileStatus, 0, fileStatus.getLen());
assertSmallerThanOneStripe(locations, dataBlocksNum);
//Test FileContext
fileStatus = fileContext.listLocatedStatus(new Path("/ec")).next();
assertSmallerThanOneStripe(fileStatus.getBlockLocations(), dataBlocksNum);
locations = fileContext.getFileBlockLocations(new Path("/ec/smallstripe"),
0, fileStatus.getLen());
assertSmallerThanOneStripe(locations, dataBlocksNum);
}
private void assertSmallerThanOneStripe(BlockLocation[] locations,
int dataBlocksNum) throws IOException {
int expectedHostNum = dataBlocksNum + parityBlocks;
assertTrue(locations.length == 1);
BlockLocation blockLocation = locations[0];
assertTrue(blockLocation.getHosts().length == expectedHostNum);
assertTrue(blockLocation.getOffset() == 0);
assertTrue(blockLocation.getLength() == dataBlocksNum * cellSize);
}
@Test(timeout=60000)
public void testListECFilesMoreThanOneBlockGroup() throws Exception {
createFile("/ec/group", blockGroupSize + 123);
RemoteIterator<LocatedFileStatus> iter =
cluster.getFileSystem().listFiles(new Path("/ec"), true);
LocatedFileStatus fileStatus = iter.next();
assertMoreThanOneBlockGroup(fileStatus.getBlockLocations(), 123);
BlockLocation[] locations = cluster.getFileSystem().getFileBlockLocations(
fileStatus, 0, fileStatus.getLen());
assertMoreThanOneBlockGroup(locations, 123);
//Test FileContext
iter = fileContext.listLocatedStatus(new Path("/ec"));
fileStatus = iter.next();
assertMoreThanOneBlockGroup(fileStatus.getBlockLocations(), 123);
locations = fileContext.getFileBlockLocations(new Path("/ec/group"),
0, fileStatus.getLen());
assertMoreThanOneBlockGroup(locations, 123);
}
private void assertMoreThanOneBlockGroup(BlockLocation[] locations,
int lastBlockSize) throws IOException {
assertTrue(locations.length == 2);
BlockLocation fistBlockGroup = locations[0];
assertTrue(fistBlockGroup.getHosts().length == numDNs);
assertTrue(fistBlockGroup.getOffset() == 0);
assertTrue(fistBlockGroup.getLength() == blockGroupSize);
BlockLocation lastBlock = locations[1];
assertTrue(lastBlock.getHosts().length == 1 + parityBlocks);
assertTrue(lastBlock.getOffset() == blockGroupSize);
assertTrue(lastBlock.getLength() == lastBlockSize);
}
}