@Test public void testSplitFile() 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); Collection<WorkUnit> splitWorkUnits = createMockSplitWorkUnits(fs, mockFileLen, mockBlockSize, mockMaxSplitSize); Assert.assertEquals(splitWorkUnits.size(), expectedSplits); Set<Integer> splitNums = new HashSet<>(); boolean hasLastSplit = false; for (WorkUnit wu : splitWorkUnits) { Optional<DistcpFileSplitter.Split> split = DistcpFileSplitter.getSplit(wu); Assert.assertTrue(split.isPresent()); Assert.assertEquals(split.get().getTotalSplits(), expectedSplits); int splitNum = split.get().getSplitNumber(); Assert.assertFalse(splitNums.contains(splitNum)); splitNums.add(splitNum); Assert.assertEquals(split.get().getLowPosition(), expectedSplitSize * splitNum); if (split.get().isLastSplit()) { hasLastSplit = true; Assert.assertEquals(split.get().getHighPosition(), mockFileLen); } else { Assert.assertEquals(split.get().getHighPosition(), expectedSplitSize * (splitNum + 1)); } } Assert.assertTrue(hasLastSplit); }
public static Collection<WorkUnit> splitFile(CopyableFile file, WorkUnit workUnit, FileSystem targetFs) throws IOException { long len = file.getFileStatus().getLen(); long blockSize = ArithmeticUtils.lcm(file.getFileStatus().getBlockSize(), file.getBlockSize(targetFs)); long maxSplitSize = workUnit.getPropAsLong(MAX_SPLIT_SIZE_KEY, DEFAULT_MAX_SPLIT_SIZE); if (maxSplitSize < blockSize) { log.warn(String.format("Max split size must be at least block size. Adjusting to %d.", blockSize)); maxSplitSize = blockSize; } if (len < maxSplitSize) { return Lists.newArrayList(workUnit); } Collection<WorkUnit> newWorkUnits = Lists.newArrayList(); long lengthPerSplit = (maxSplitSize / blockSize) * blockSize; int splits = (int) (len / lengthPerSplit + 1); for (int i = 0; i < splits; i++) { WorkUnit newWorkUnit = WorkUnit.copyOf(workUnit); long lowPos = lengthPerSplit * i; long highPos = Math.min(lengthPerSplit * (i + 1), len); Split split = new Split(lowPos, highPos, i, splits, String.format("%s.__PART%d__", file.getDestination().getName(), i)); String serializedSplit = GSON.toJson(split); newWorkUnit.setProp(SPLIT_KEY, serializedSplit); Guid oldGuid = CopySource.getWorkUnitGuid(newWorkUnit).get(); Guid newGuid = oldGuid.append(Guid.fromStrings(serializedSplit)); CopySource.setWorkUnitGuid(workUnit, newGuid); newWorkUnits.add(newWorkUnit); } return newWorkUnits; }
DistcpFileSplitter { public static Collection<WorkUnit> splitFile(CopyableFile file, WorkUnit workUnit, FileSystem targetFs) throws IOException { long len = file.getFileStatus().getLen(); long blockSize = ArithmeticUtils.lcm(file.getFileStatus().getBlockSize(), file.getBlockSize(targetFs)); long maxSplitSize = workUnit.getPropAsLong(MAX_SPLIT_SIZE_KEY, DEFAULT_MAX_SPLIT_SIZE); if (maxSplitSize < blockSize) { log.warn(String.format("Max split size must be at least block size. Adjusting to %d.", blockSize)); maxSplitSize = blockSize; } if (len < maxSplitSize) { return Lists.newArrayList(workUnit); } Collection<WorkUnit> newWorkUnits = Lists.newArrayList(); long lengthPerSplit = (maxSplitSize / blockSize) * blockSize; int splits = (int) (len / lengthPerSplit + 1); for (int i = 0; i < splits; i++) { WorkUnit newWorkUnit = WorkUnit.copyOf(workUnit); long lowPos = lengthPerSplit * i; long highPos = Math.min(lengthPerSplit * (i + 1), len); Split split = new Split(lowPos, highPos, i, splits, String.format("%s.__PART%d__", file.getDestination().getName(), i)); String serializedSplit = GSON.toJson(split); newWorkUnit.setProp(SPLIT_KEY, serializedSplit); Guid oldGuid = CopySource.getWorkUnitGuid(newWorkUnit).get(); Guid newGuid = oldGuid.append(Guid.fromStrings(serializedSplit)); CopySource.setWorkUnitGuid(workUnit, newGuid); newWorkUnits.add(newWorkUnit); } return newWorkUnits; } }
DistcpFileSplitter { public static Collection<WorkUnit> splitFile(CopyableFile file, WorkUnit workUnit, FileSystem targetFs) throws IOException { long len = file.getFileStatus().getLen(); long blockSize = ArithmeticUtils.lcm(file.getFileStatus().getBlockSize(), file.getBlockSize(targetFs)); long maxSplitSize = workUnit.getPropAsLong(MAX_SPLIT_SIZE_KEY, DEFAULT_MAX_SPLIT_SIZE); if (maxSplitSize < blockSize) { log.warn(String.format("Max split size must be at least block size. Adjusting to %d.", blockSize)); maxSplitSize = blockSize; } if (len < maxSplitSize) { return Lists.newArrayList(workUnit); } Collection<WorkUnit> newWorkUnits = Lists.newArrayList(); long lengthPerSplit = (maxSplitSize / blockSize) * blockSize; int splits = (int) (len / lengthPerSplit + 1); for (int i = 0; i < splits; i++) { WorkUnit newWorkUnit = WorkUnit.copyOf(workUnit); long lowPos = lengthPerSplit * i; long highPos = Math.min(lengthPerSplit * (i + 1), len); Split split = new Split(lowPos, highPos, i, splits, String.format("%s.__PART%d__", file.getDestination().getName(), i)); String serializedSplit = GSON.toJson(split); newWorkUnit.setProp(SPLIT_KEY, serializedSplit); Guid oldGuid = CopySource.getWorkUnitGuid(newWorkUnit).get(); Guid newGuid = oldGuid.append(Guid.fromStrings(serializedSplit)); CopySource.setWorkUnitGuid(workUnit, newGuid); newWorkUnits.add(newWorkUnit); } return newWorkUnits; }  }
DistcpFileSplitter { public static Collection<WorkUnit> splitFile(CopyableFile file, WorkUnit workUnit, FileSystem targetFs) throws IOException { long len = file.getFileStatus().getLen(); long blockSize = ArithmeticUtils.lcm(file.getFileStatus().getBlockSize(), file.getBlockSize(targetFs)); long maxSplitSize = workUnit.getPropAsLong(MAX_SPLIT_SIZE_KEY, DEFAULT_MAX_SPLIT_SIZE); if (maxSplitSize < blockSize) { log.warn(String.format("Max split size must be at least block size. Adjusting to %d.", blockSize)); maxSplitSize = blockSize; } if (len < maxSplitSize) { return Lists.newArrayList(workUnit); } Collection<WorkUnit> newWorkUnits = Lists.newArrayList(); long lengthPerSplit = (maxSplitSize / blockSize) * blockSize; int splits = (int) (len / lengthPerSplit + 1); for (int i = 0; i < splits; i++) { WorkUnit newWorkUnit = WorkUnit.copyOf(workUnit); long lowPos = lengthPerSplit * i; long highPos = Math.min(lengthPerSplit * (i + 1), len); Split split = new Split(lowPos, highPos, i, splits, String.format("%s.__PART%d__", file.getDestination().getName(), i)); String serializedSplit = GSON.toJson(split); newWorkUnit.setProp(SPLIT_KEY, serializedSplit); Guid oldGuid = CopySource.getWorkUnitGuid(newWorkUnit).get(); Guid newGuid = oldGuid.append(Guid.fromStrings(serializedSplit)); CopySource.setWorkUnitGuid(workUnit, newGuid); newWorkUnits.add(newWorkUnit); } return newWorkUnits; }  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<WorkUnit> splitFile(CopyableFile file, WorkUnit workUnit, FileSystem targetFs) throws IOException { long len = file.getFileStatus().getLen(); long blockSize = ArithmeticUtils.lcm(file.getFileStatus().getBlockSize(), file.getBlockSize(targetFs)); long maxSplitSize = workUnit.getPropAsLong(MAX_SPLIT_SIZE_KEY, DEFAULT_MAX_SPLIT_SIZE); if (maxSplitSize < blockSize) { log.warn(String.format("Max split size must be at least block size. Adjusting to %d.", blockSize)); maxSplitSize = blockSize; } if (len < maxSplitSize) { return Lists.newArrayList(workUnit); } Collection<WorkUnit> newWorkUnits = Lists.newArrayList(); long lengthPerSplit = (maxSplitSize / blockSize) * blockSize; int splits = (int) (len / lengthPerSplit + 1); for (int i = 0; i < splits; i++) { WorkUnit newWorkUnit = WorkUnit.copyOf(workUnit); long lowPos = lengthPerSplit * i; long highPos = Math.min(lengthPerSplit * (i + 1), len); Split split = new Split(lowPos, highPos, i, splits, String.format("%s.__PART%d__", file.getDestination().getName(), i)); String serializedSplit = GSON.toJson(split); newWorkUnit.setProp(SPLIT_KEY, serializedSplit); Guid oldGuid = CopySource.getWorkUnitGuid(newWorkUnit).get(); Guid newGuid = oldGuid.append(Guid.fromStrings(serializedSplit)); CopySource.setWorkUnitGuid(workUnit, newGuid); newWorkUnits.add(newWorkUnit); } return newWorkUnits; }  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; }