@Test public void testMergeSplitWorkUnits() throws Exception { long mockFileLen = 12L; long mockBlockSize = 4L; long mockMaxSplitSize = 4L; long expectedSplitSize = (mockMaxSplitSize / mockBlockSize) * mockBlockSize; int expectedSplits = (int) (mockFileLen / expectedSplitSize + 1); FileSystem fs = mock(FileSystem.class); List<WorkUnitState> splitWorkUnits = createMockSplitWorkUnits(fs, mockFileLen, mockBlockSize, mockMaxSplitSize).stream() .map(wu -> new WorkUnitState(wu)).collect(Collectors.toList()); Assert.assertEquals(splitWorkUnits.size(), expectedSplits); Collection<WorkUnitState> mergedWorkUnits = DistcpFileSplitter.mergeAllSplitWorkUnits(fs, splitWorkUnits); Assert.assertEquals(mergedWorkUnits.size(), 1); }
public static Collection<WorkUnitState> mergeAllSplitWorkUnits(FileSystem fs, Collection<WorkUnitState> workUnits) throws IOException { ListMultimap<CopyableFile, WorkUnitState> splitWorkUnitsMap = ArrayListMultimap.create(); for (WorkUnitState workUnit : workUnits) { if (isSplitWorkUnit(workUnit)) { CopyableFile copyableFile = (CopyableFile) CopySource.deserializeCopyEntity(workUnit); splitWorkUnitsMap.put(copyableFile, workUnit); } } for (CopyableFile file : splitWorkUnitsMap.keySet()) { log.info(String.format("Merging split file %s.", file.getDestination())); WorkUnitState oldWorkUnit = splitWorkUnitsMap.get(file).get(0); Path outputDir = FileAwareInputStreamDataWriter.getOutputDir(oldWorkUnit); CopyEntity.DatasetAndPartition datasetAndPartition = file.getDatasetAndPartition(CopySource.deserializeCopyableDataset(oldWorkUnit)); Path parentPath = FileAwareInputStreamDataWriter.getOutputFilePath(file, outputDir, datasetAndPartition) .getParent(); WorkUnitState newWorkUnit = mergeSplits(fs, file, splitWorkUnitsMap.get(file), parentPath); for (WorkUnitState wu : splitWorkUnitsMap.get(file)) { wu.setWorkingState(WorkUnitState.WorkingState.COMMITTED); workUnits.remove(wu); } workUnits.add(newWorkUnit); } return workUnits; }
DistcpFileSplitter { public static Collection<WorkUnitState> mergeAllSplitWorkUnits(FileSystem fs, Collection<WorkUnitState> workUnits) throws IOException { ListMultimap<CopyableFile, WorkUnitState> splitWorkUnitsMap = ArrayListMultimap.create(); for (WorkUnitState workUnit : workUnits) { if (isSplitWorkUnit(workUnit)) { CopyableFile copyableFile = (CopyableFile) CopySource.deserializeCopyEntity(workUnit); splitWorkUnitsMap.put(copyableFile, workUnit); } } for (CopyableFile file : splitWorkUnitsMap.keySet()) { log.info(String.format("Merging split file %s.", file.getDestination())); WorkUnitState oldWorkUnit = splitWorkUnitsMap.get(file).get(0); Path outputDir = FileAwareInputStreamDataWriter.getOutputDir(oldWorkUnit); CopyEntity.DatasetAndPartition datasetAndPartition = file.getDatasetAndPartition(CopySource.deserializeCopyableDataset(oldWorkUnit)); Path parentPath = FileAwareInputStreamDataWriter.getOutputFilePath(file, outputDir, datasetAndPartition) .getParent(); WorkUnitState newWorkUnit = mergeSplits(fs, file, splitWorkUnitsMap.get(file), parentPath); for (WorkUnitState wu : splitWorkUnitsMap.get(file)) { wu.setWorkingState(WorkUnitState.WorkingState.COMMITTED); workUnits.remove(wu); } workUnits.add(newWorkUnit); } return workUnits; } }
DistcpFileSplitter { public static Collection<WorkUnitState> mergeAllSplitWorkUnits(FileSystem fs, Collection<WorkUnitState> workUnits) throws IOException { ListMultimap<CopyableFile, WorkUnitState> splitWorkUnitsMap = ArrayListMultimap.create(); for (WorkUnitState workUnit : workUnits) { if (isSplitWorkUnit(workUnit)) { CopyableFile copyableFile = (CopyableFile) CopySource.deserializeCopyEntity(workUnit); splitWorkUnitsMap.put(copyableFile, workUnit); } } for (CopyableFile file : splitWorkUnitsMap.keySet()) { log.info(String.format("Merging split file %s.", file.getDestination())); WorkUnitState oldWorkUnit = splitWorkUnitsMap.get(file).get(0); Path outputDir = FileAwareInputStreamDataWriter.getOutputDir(oldWorkUnit); CopyEntity.DatasetAndPartition datasetAndPartition = file.getDatasetAndPartition(CopySource.deserializeCopyableDataset(oldWorkUnit)); Path parentPath = FileAwareInputStreamDataWriter.getOutputFilePath(file, outputDir, datasetAndPartition) .getParent(); WorkUnitState newWorkUnit = mergeSplits(fs, file, splitWorkUnitsMap.get(file), parentPath); for (WorkUnitState wu : splitWorkUnitsMap.get(file)) { wu.setWorkingState(WorkUnitState.WorkingState.COMMITTED); workUnits.remove(wu); } workUnits.add(newWorkUnit); } return workUnits; }  }
DistcpFileSplitter { public static Collection<WorkUnitState> mergeAllSplitWorkUnits(FileSystem fs, Collection<WorkUnitState> workUnits) throws IOException { ListMultimap<CopyableFile, WorkUnitState> splitWorkUnitsMap = ArrayListMultimap.create(); for (WorkUnitState workUnit : workUnits) { if (isSplitWorkUnit(workUnit)) { CopyableFile copyableFile = (CopyableFile) CopySource.deserializeCopyEntity(workUnit); splitWorkUnitsMap.put(copyableFile, workUnit); } } for (CopyableFile file : splitWorkUnitsMap.keySet()) { log.info(String.format("Merging split file %s.", file.getDestination())); WorkUnitState oldWorkUnit = splitWorkUnitsMap.get(file).get(0); Path outputDir = FileAwareInputStreamDataWriter.getOutputDir(oldWorkUnit); CopyEntity.DatasetAndPartition datasetAndPartition = file.getDatasetAndPartition(CopySource.deserializeCopyableDataset(oldWorkUnit)); Path parentPath = FileAwareInputStreamDataWriter.getOutputFilePath(file, outputDir, datasetAndPartition) .getParent(); WorkUnitState newWorkUnit = mergeSplits(fs, file, splitWorkUnitsMap.get(file), parentPath); for (WorkUnitState wu : splitWorkUnitsMap.get(file)) { wu.setWorkingState(WorkUnitState.WorkingState.COMMITTED); workUnits.remove(wu); } workUnits.add(newWorkUnit); } return workUnits; }  static Collection<WorkUnit> splitFile(CopyableFile file, WorkUnit workUnit, FileSystem targetFs); static Collection<WorkUnitState> mergeAllSplitWorkUnits(FileSystem fs, Collection<WorkUnitState> workUnits); static boolean isSplitWorkUnit(State workUnit); static Optional<Split> getSplit(State workUnit); static boolean allowSplit(State state, FileSystem targetFs); }
DistcpFileSplitter { public static Collection<WorkUnitState> mergeAllSplitWorkUnits(FileSystem fs, Collection<WorkUnitState> workUnits) throws IOException { ListMultimap<CopyableFile, WorkUnitState> splitWorkUnitsMap = ArrayListMultimap.create(); for (WorkUnitState workUnit : workUnits) { if (isSplitWorkUnit(workUnit)) { CopyableFile copyableFile = (CopyableFile) CopySource.deserializeCopyEntity(workUnit); splitWorkUnitsMap.put(copyableFile, workUnit); } } for (CopyableFile file : splitWorkUnitsMap.keySet()) { log.info(String.format("Merging split file %s.", file.getDestination())); WorkUnitState oldWorkUnit = splitWorkUnitsMap.get(file).get(0); Path outputDir = FileAwareInputStreamDataWriter.getOutputDir(oldWorkUnit); CopyEntity.DatasetAndPartition datasetAndPartition = file.getDatasetAndPartition(CopySource.deserializeCopyableDataset(oldWorkUnit)); Path parentPath = FileAwareInputStreamDataWriter.getOutputFilePath(file, outputDir, datasetAndPartition) .getParent(); WorkUnitState newWorkUnit = mergeSplits(fs, file, splitWorkUnitsMap.get(file), parentPath); for (WorkUnitState wu : splitWorkUnitsMap.get(file)) { wu.setWorkingState(WorkUnitState.WorkingState.COMMITTED); workUnits.remove(wu); } workUnits.add(newWorkUnit); } return workUnits; }  static Collection<WorkUnit> splitFile(CopyableFile file, WorkUnit workUnit, FileSystem targetFs); static Collection<WorkUnitState> mergeAllSplitWorkUnits(FileSystem fs, Collection<WorkUnitState> workUnits); static boolean isSplitWorkUnit(State workUnit); static Optional<Split> getSplit(State workUnit); static boolean allowSplit(State state, FileSystem targetFs); static final String SPLIT_ENABLED; static final String MAX_SPLIT_SIZE_KEY; static final long DEFAULT_MAX_SPLIT_SIZE; static final Set<String> KNOWN_SCHEMES_SUPPORTING_CONCAT; }