big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MAPREDUCE-5014. Extend Distcp to accept a custom CopyListing. Contributed by Srikanth Sundarrajan

Browse Source 
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1459690 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Amareshwari Sri Ramadasu 2013-03-22 10:09:06 +00:00
parent 92bee12fac
commit 6f92ad9868
7 changed files with 171 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-mapreduce-project/CHANGES.txt
View File

@ -75,6 +75,9 @@ Trunk (Unreleased)
MAPREDUCE-4735. Make arguments in TestDFSIO case insensitive.
(Brandon Li via suresh)
MAPREDUCE-5014. Extend Distcp to accept a custom CopyListing.
(Srikanth Sundarrajan via amareshwari)
BUG FIXES
MAPREDUCE-4272. SortedRanges.Range#compareTo is not spec compliant.

31
hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/CopyListing.java
View File

@ -30,6 +30,7 @@
import org.apache.hadoop.security.Credentials;
import java.io.IOException;
import java.lang.reflect.Constructor;
/**
* The CopyListing abstraction is responsible for how the list of
@ -193,14 +194,34 @@ protected Credentials getCredentials() {
* @param credentials Credentials object on which the FS delegation tokens are cached
* @param options The input Options, to help choose the appropriate CopyListing Implementation.
* @return An instance of the appropriate CopyListing implementation.
* @throws java.io.IOException - Exception if any
*/
public static CopyListing getCopyListing(Configuration configuration,
Credentials credentials,
DistCpOptions options) {
if (options.getSourceFileListing() == null) {
return new GlobbedCopyListing(configuration, credentials);
} else {
return new FileBasedCopyListing(configuration, credentials);
DistCpOptions options)
throws IOException {
String copyListingClassName = configuration.get(DistCpConstants.
CONF_LABEL_COPY_LISTING_CLASS, "");
Class<? extends CopyListing> copyListingClass;
try {
if (! copyListingClassName.isEmpty()) {
copyListingClass = configuration.getClass(DistCpConstants.
CONF_LABEL_COPY_LISTING_CLASS, GlobbedCopyListing.class,
CopyListing.class);
} else {
if (options.getSourceFileListing() == null) {
copyListingClass = GlobbedCopyListing.class;
} else {
copyListingClass = FileBasedCopyListing.class;
}
}
copyListingClassName = copyListingClass.getName();
Constructor<? extends CopyListing> constructor = copyListingClass.
getDeclaredConstructor(Configuration.class, Credentials.class);
return constructor.newInstance(configuration, credentials);
} catch (Exception e) {
throw new IOException("Unable to instantiate " + copyListingClassName, e);
}
}

4
hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCp.java
View File

@ -319,7 +319,7 @@ private void configureOutputFormat(Job job) throws IOException {
* @return Returns the path where the copy listing is created
* @throws IOException - If any
*/
private Path createInputFileListing(Job job) throws IOException {
protected Path createInputFileListing(Job job) throws IOException {
Path fileListingPath = getFileListingPath();
CopyListing copyListing = CopyListing.getCopyListing(job.getConfiguration(),
job.getCredentials(), inputOptions);
@ -334,7 +334,7 @@ private Path createInputFileListing(Job job) throws IOException {
* @return - Path where the copy listing file has to be saved
* @throws IOException - Exception if any
*/
private Path getFileListingPath() throws IOException {
protected Path getFileListingPath() throws IOException {
String fileListPathStr = metaFolder + "/fileList.seq";
Path path = new Path(fileListPathStr);
return new Path(path.toUri().normalize().toString());

3
hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpConstants.java
View File

@ -82,6 +82,9 @@ public class DistCpConstants {
/* Meta folder where the job's intermediate data is kept */
public static final String CONF_LABEL_META_FOLDER = "distcp.meta.folder";
/* DistCp CopyListing class override param */
public static final String CONF_LABEL_COPY_LISTING_CLASS = "distcp.copy.listing.class";
/**
* Conf label for SSL Trust-store location.
*/

37
hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/SimpleCopyListing.java
View File

@ -127,17 +127,20 @@ public void doBuildListing(Path pathToListingFile, DistCpOptions options) throws
if (LOG.isDebugEnabled()) {
LOG.debug("Recording source-path: " + sourceStatus.getPath() + " for copy.");
}
writeToFileListing(fileListWriter, sourceStatus, sourcePathRoot, localFile);
writeToFileListing(fileListWriter, sourceStatus, sourcePathRoot,
localFile, options);
if (isDirectoryAndNotEmpty(sourceFS, sourceStatus)) {
if (LOG.isDebugEnabled()) {
LOG.debug("Traversing non-empty source dir: " + sourceStatus.getPath());
}
traverseNonEmptyDirectory(fileListWriter, sourceStatus, sourcePathRoot, localFile);
traverseNonEmptyDirectory(fileListWriter, sourceStatus, sourcePathRoot,
localFile, options);
}
}
} else {
writeToFileListing(fileListWriter, rootStatus, sourcePathRoot, localFile);
writeToFileListing(fileListWriter, rootStatus, sourcePathRoot,
localFile, options);
}
}
} finally {
@ -169,6 +172,17 @@ private Path computeSourceRootPath(FileStatus sourceStatus,
}
}
/**
* Provide an option to skip copy of a path, Allows for exclusion
* of files such as {@link org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter#SUCCEEDED_FILE_NAME}
* @param path - Path being considered for copy while building the file listing
* @param options - Input options passed during DistCp invocation
* @return - True if the path should be considered for copy, false otherwise
*/
protected boolean shouldCopy(Path path, DistCpOptions options) {
return true;
}
/** {@inheritDoc} */
@Override
protected long getBytesToCopy() {
@ -210,7 +224,9 @@ private static FileStatus[] getChildren(FileSystem fileSystem,
private void traverseNonEmptyDirectory(SequenceFile.Writer fileListWriter,
FileStatus sourceStatus,
Path sourcePathRoot, boolean localFile)
Path sourcePathRoot,
boolean localFile,
DistCpOptions options)
throws IOException {
FileSystem sourceFS = sourcePathRoot.getFileSystem(getConf());
Stack<FileStatus> pathStack = new Stack<FileStatus>();
@ -221,7 +237,8 @@ private void traverseNonEmptyDirectory(SequenceFile.Writer fileListWriter,
if (LOG.isDebugEnabled())
LOG.debug("Recording source-path: "
+ sourceStatus.getPath() + " for copy.");
writeToFileListing(fileListWriter, child, sourcePathRoot, localFile);
writeToFileListing(fileListWriter, child, sourcePathRoot,
localFile, options);
if (isDirectoryAndNotEmpty(sourceFS, child)) {
if (LOG.isDebugEnabled())
LOG.debug("Traversing non-empty source dir: "
@ -233,8 +250,10 @@ private void traverseNonEmptyDirectory(SequenceFile.Writer fileListWriter,
}
private void writeToFileListing(SequenceFile.Writer fileListWriter,
FileStatus fileStatus, Path sourcePathRoot,
boolean localFile) throws IOException {
FileStatus fileStatus,
Path sourcePathRoot,
boolean localFile,
DistCpOptions options) throws IOException {
if (fileStatus.getPath().equals(sourcePathRoot) && fileStatus.isDirectory())
return; // Skip the root-paths.
@ -248,6 +267,10 @@ private void writeToFileListing(SequenceFile.Writer fileListWriter,
status = getFileStatus(fileStatus);
}
if (!shouldCopy(fileStatus.getPath(), options)) {
return;
}
fileListWriter.append(new Text(DistCpUtils.getRelativePath(sourcePathRoot,
fileStatus.getPath())), status);
fileListWriter.sync();

41
hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestCopyListing.java
View File

@ -24,6 +24,7 @@
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.tools.util.TestDistCpUtils;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.security.Credentials;
@ -79,7 +80,39 @@ protected long getNumberOfPaths() {
return 0;
}
@Test
@Test(timeout=10000)
public void testSkipCopy() throws Exception {
SimpleCopyListing listing = new SimpleCopyListing(getConf(), CREDENTIALS) {
@Override
protected boolean shouldCopy(Path path, DistCpOptions options) {
return !path.getName().equals(FileOutputCommitter.SUCCEEDED_FILE_NAME);
}
};
FileSystem fs = FileSystem.get(getConf());
List<Path> srcPaths = new ArrayList<Path>();
srcPaths.add(new Path("/tmp/in4/1"));
srcPaths.add(new Path("/tmp/in4/2"));
Path target = new Path("/tmp/out4/1");
TestDistCpUtils.createFile(fs, "/tmp/in4/1/_SUCCESS");
TestDistCpUtils.createFile(fs, "/tmp/in4/1/file");
TestDistCpUtils.createFile(fs, "/tmp/in4/2");
fs.mkdirs(target);
DistCpOptions options = new DistCpOptions(srcPaths, target);
Path listingFile = new Path("/tmp/list4");
listing.buildListing(listingFile, options);
Assert.assertEquals(listing.getNumberOfPaths(), 2);
SequenceFile.Reader reader = new SequenceFile.Reader(getConf(),
SequenceFile.Reader.file(listingFile));
FileStatus fileStatus = new FileStatus();
Text relativePath = new Text();
Assert.assertTrue(reader.next(relativePath, fileStatus));
Assert.assertEquals(relativePath.toString(), "/1/file");
Assert.assertTrue(reader.next(relativePath, fileStatus));
Assert.assertEquals(relativePath.toString(), "/2");
Assert.assertFalse(reader.next(relativePath, fileStatus));
}
@Test(timeout=10000)
public void testMultipleSrcToFile() {
FileSystem fs = null;
try {
@ -124,7 +157,7 @@ public void testMultipleSrcToFile() {
}
}
@Test
@Test(timeout=10000)
public void testDuplicates() {
FileSystem fs = null;
try {
@ -150,7 +183,7 @@ public void testDuplicates() {
}
}
@Test
@Test(timeout=10000)
public void testBuildListing() {
FileSystem fs = null;
try {
@ -206,7 +239,7 @@ public void testBuildListing() {
}
}
@Test
@Test(timeout=10000)
public void testBuildListingForSingleFile() {
FileSystem fs = null;
String testRootString = "/singleFileListing";

89
hadoop-tools/hadoop-distcp/src/test/java/org/apache/hadoop/tools/TestIntegration.java
View File

@ -26,6 +26,7 @@
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Cluster;
import org.apache.hadoop.mapreduce.JobSubmissionFiles;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.tools.util.TestDistCpUtils;
import org.junit.Assert;
import org.junit.BeforeClass;
@ -34,6 +35,7 @@
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class TestIntegration {
@ -68,7 +70,7 @@ public static void setup() {
}
}
@Test
@Test(timeout=100000)
public void testSingleFileMissingTarget() {
caseSingleFileMissingTarget(false);
caseSingleFileMissingTarget(true);
@ -91,7 +93,7 @@ private void caseSingleFileMissingTarget(boolean sync) {
}
}
@Test
@Test(timeout=100000)
public void testSingleFileTargetFile() {
caseSingleFileTargetFile(false);
caseSingleFileTargetFile(true);
@ -114,7 +116,7 @@ private void caseSingleFileTargetFile(boolean sync) {
}
}
@Test
@Test(timeout=100000)
public void testSingleFileTargetDir() {
caseSingleFileTargetDir(false);
caseSingleFileTargetDir(true);
@ -138,7 +140,7 @@ private void caseSingleFileTargetDir(boolean sync) {
}
}
@Test
@Test(timeout=100000)
public void testSingleDirTargetMissing() {
caseSingleDirTargetMissing(false);
caseSingleDirTargetMissing(true);
@ -161,7 +163,7 @@ private void caseSingleDirTargetMissing(boolean sync) {
}
}
@Test
@Test(timeout=100000)
public void testSingleDirTargetPresent() {
try {
@ -180,7 +182,7 @@ public void testSingleDirTargetPresent() {
}
}
@Test
@Test(timeout=100000)
public void testUpdateSingleDirTargetPresent() {
try {
@ -199,7 +201,7 @@ public void testUpdateSingleDirTargetPresent() {
}
}
@Test
@Test(timeout=100000)
public void testMultiFileTargetPresent() {
caseMultiFileTargetPresent(false);
caseMultiFileTargetPresent(true);
@ -223,7 +225,56 @@ private void caseMultiFileTargetPresent(boolean sync) {
}
}
@Test
@Test(timeout=100000)
public void testCustomCopyListing() {
try {
addEntries(listFile, "multifile1/file3", "multifile1/file4", "multifile1/file5");
createFiles("multifile1/file3", "multifile1/file4", "multifile1/file5");
mkdirs(target.toString());
Configuration conf = getConf();
try {
conf.setClass(DistCpConstants.CONF_LABEL_COPY_LISTING_CLASS,
CustomCopyListing.class, CopyListing.class);
DistCpOptions options = new DistCpOptions(Arrays.
asList(new Path(root + "/" + "multifile1")), target);
options.setSyncFolder(true);
options.setDeleteMissing(false);
options.setOverwrite(false);
try {
new DistCp(conf, options).execute();
} catch (Exception e) {
LOG.error("Exception encountered ", e);
throw new IOException(e);
}
} finally {
conf.unset(DistCpConstants.CONF_LABEL_COPY_LISTING_CLASS);
}
checkResult(target, 2, "file4", "file5");
} catch (IOException e) {
LOG.error("Exception encountered while testing distcp", e);
Assert.fail("distcp failure");
} finally {
TestDistCpUtils.delete(fs, root);
}
}
private static class CustomCopyListing extends SimpleCopyListing {
public CustomCopyListing(Configuration configuration,
Credentials credentials) {
super(configuration, credentials);
}
@Override
protected boolean shouldCopy(Path path, DistCpOptions options) {
return !path.getName().equals("file3");
}
}
@Test(timeout=100000)
public void testMultiFileTargetMissing() {
caseMultiFileTargetMissing(false);
caseMultiFileTargetMissing(true);
@ -246,7 +297,7 @@ private void caseMultiFileTargetMissing(boolean sync) {
}
}
@Test
@Test(timeout=100000)
public void testMultiDirTargetPresent() {
try {
@ -265,7 +316,7 @@ public void testMultiDirTargetPresent() {
}
}
@Test
@Test(timeout=100000)
public void testUpdateMultiDirTargetPresent() {
try {
@ -284,7 +335,7 @@ public void testUpdateMultiDirTargetPresent() {
}
}
@Test
@Test(timeout=100000)
public void testMultiDirTargetMissing() {
try {
@ -304,7 +355,7 @@ public void testMultiDirTargetMissing() {
}
}
@Test
@Test(timeout=100000)
public void testUpdateMultiDirTargetMissing() {
try {
@ -323,7 +374,7 @@ public void testUpdateMultiDirTargetMissing() {
}
}
@Test
@Test(timeout=100000)
public void testDeleteMissingInDestination() {
try {
@ -343,7 +394,7 @@ public void testDeleteMissingInDestination() {
}
}
@Test
@Test(timeout=100000)
public void testOverwrite() {
byte[] contents1 = "contents1".getBytes();
byte[] contents2 = "contents2".getBytes();
@ -375,7 +426,7 @@ public void testOverwrite() {
}
}
@Test
@Test(timeout=100000)
public void testGlobTargetMissingSingleLevel() {
try {
@ -398,7 +449,7 @@ public void testGlobTargetMissingSingleLevel() {
}
}
@Test
@Test(timeout=100000)
public void testUpdateGlobTargetMissingSingleLevel() {
try {
@ -420,7 +471,7 @@ public void testUpdateGlobTargetMissingSingleLevel() {
}
}
@Test
@Test(timeout=100000)
public void testGlobTargetMissingMultiLevel() {
try {
@ -444,7 +495,7 @@ public void testGlobTargetMissingMultiLevel() {
}
}
@Test
@Test(timeout=100000)
public void testUpdateGlobTargetMissingMultiLevel() {
try {
@ -468,7 +519,7 @@ public void testUpdateGlobTargetMissingMultiLevel() {
}
}
@Test
@Test(timeout=100000)
public void testCleanup() {
try {
Path sourcePath = new Path("noscheme:///file");