DistCp extends Configured implements Tool { public Job execute() throws Exception { assert inputOptions != null; assert getConf() != null; Job job = null; try { metaFolder = createMetaFolderPath(); jobFS = metaFolder.getFileSystem(getConf()); job = createJob(); createInputFileListing(job); job.submit(); submitted = true; } finally { if (!submitted) { cleanup(); } } String jobID = getJobID(job); job.getConfiguration().set(DistCpConstants.CONF_LABEL_DISTCP_JOB_ID, jobID); LOG.info("DistCp job-id: " + jobID); LOG.info("DistCp job may be tracked at: " + job.getTrackingURL()); LOG.info("To cancel, run the following command:\thadoop job -kill " + jobID); long jobStartTime = System.nanoTime(); if (inputOptions.shouldBlock() && !job.waitForCompletion(true)) { updateJobTimeInNanos(jobStartTime); throw new IOException("DistCp failure: Job " + jobID + " has failed. "); } updateJobTimeInNanos(jobStartTime); return job; } DistCp(Configuration configuration, DistCpOptions inputOptions); private  DistCp(); int run(String[] argv); Job execute(); long getJobTimeInNanos(); static void main(String argv[]); static final Random rand; }
@Test public void testRootPath() throws Exception { try { clearState(); List<Path> sources = new ArrayList<Path>(); sources.add(new Path("/a") .makeQualified(cluster.getFileSystem())); sources.add(new Path("/b") .makeQualified(cluster.getFileSystem())); touchFile("/a/a.gz"); touchFile("/b/b.gz"); Path targetPath = new Path("/c") .makeQualified(cluster.getFileSystem()); DistCpOptions options = new DistCpOptions(sources, targetPath); options.setOutPutDirectory(counterOutputPath); new DistCp(configuration, options).execute(); Assert.assertTrue(cluster.getFileSystem().exists(new Path("/c/a/a.gz"))); Assert.assertTrue(cluster.getFileSystem().exists(new Path("/c/b/b.gz"))); } catch (Exception e) { LOG.error("Exception encountered", e); Assert.fail("Unexpected exception: " + e.getMessage()); } }
@Test public void testDynamicDistCp() throws Exception { try { clearState(); Path sourcePath = new Path(SOURCE_PATH) .makeQualified(cluster.getFileSystem()); List<Path> sources = new ArrayList<Path>(); sources.add(sourcePath); Path targetPath = new Path(TARGET_PATH) .makeQualified(cluster.getFileSystem()); DistCpOptions options = new DistCpOptions(sources, targetPath); options.setCopyStrategy("dynamic"); options.setOutPutDirectory(counterOutputPath); options.setAtomicCommit(true); options.setAtomicWorkPath(new Path("/work")); options.setBlocking(false); Job job = new DistCp(configuration, options).execute(); Path workDir = CopyOutputFormat.getWorkingDirectory(job); Path finalDir = CopyOutputFormat.getCommitDirectory(job); while (!job.isComplete()) { if (cluster.getFileSystem().exists(workDir)) { break; } } job.waitForCompletion(true); Assert.assertFalse(cluster.getFileSystem().exists(workDir)); Assert.assertTrue(cluster.getFileSystem().exists(finalDir)); verifyResults(); } catch (Exception e) { LOG.error("Exception encountered", e); Assert.fail("Unexpected exception: " + e.getMessage()); } }