@Test public void splits_unaligned() { BlockMap blocks = blocks("testing", TemporaryFile.BLOCK_SIZE - 10, TemporaryFile.BLOCK_SIZE); List<FileSplit> splits = TemporaryInputFormat.computeSplits(new Path("testing"), blocks, m(128)); assertThat(splits, hasSize(2)); FileSplit s0 = find(splits, 0); assertThat(s0.getLength(), is((long) TemporaryFile.BLOCK_SIZE)); FileSplit s1 = find(splits, TemporaryFile.BLOCK_SIZE); assertThat(s1.getLength(), is((long) TemporaryFile.BLOCK_SIZE - 10)); }
static List<FileSplit> computeSplits(Path path, BlockMap blockMap, long splitSize) { long align = splitSize; if (splitSize > 0) { long remain = splitSize % TemporaryFile.BLOCK_SIZE; if (remain != 0) { align += TemporaryFile.BLOCK_SIZE - remain; } } long size = blockMap.getFileSize(); long start = 0; List<FileSplit> results = new ArrayList<>(); for (BlockInfo block : blockMap.getBlocks()) { assert start % TemporaryFile.BLOCK_SIZE == 0; long end = block.getEnd(); if (end < start) { continue; } long remain = end % TemporaryFile.BLOCK_SIZE; if (remain != 0) { end = Math.min(size, end + (TemporaryFile.BLOCK_SIZE - remain)); } results.addAll(createSplits(path, blockMap, start, end, align)); start = end; } return results; }
TemporaryInputFormat extends InputFormat<NullWritable, T> { static List<FileSplit> computeSplits(Path path, BlockMap blockMap, long splitSize) { long align = splitSize; if (splitSize > 0) { long remain = splitSize % TemporaryFile.BLOCK_SIZE; if (remain != 0) { align += TemporaryFile.BLOCK_SIZE - remain; } } long size = blockMap.getFileSize(); long start = 0; List<FileSplit> results = new ArrayList<>(); for (BlockInfo block : blockMap.getBlocks()) { assert start % TemporaryFile.BLOCK_SIZE == 0; long end = block.getEnd(); if (end < start) { continue; } long remain = end % TemporaryFile.BLOCK_SIZE; if (remain != 0) { end = Math.min(size, end + (TemporaryFile.BLOCK_SIZE - remain)); } results.addAll(createSplits(path, blockMap, start, end, align)); start = end; } return results; } }
TemporaryInputFormat extends InputFormat<NullWritable, T> { static List<FileSplit> computeSplits(Path path, BlockMap blockMap, long splitSize) { long align = splitSize; if (splitSize > 0) { long remain = splitSize % TemporaryFile.BLOCK_SIZE; if (remain != 0) { align += TemporaryFile.BLOCK_SIZE - remain; } } long size = blockMap.getFileSize(); long start = 0; List<FileSplit> results = new ArrayList<>(); for (BlockInfo block : blockMap.getBlocks()) { assert start % TemporaryFile.BLOCK_SIZE == 0; long end = block.getEnd(); if (end < start) { continue; } long remain = end % TemporaryFile.BLOCK_SIZE; if (remain != 0) { end = Math.min(size, end + (TemporaryFile.BLOCK_SIZE - remain)); } results.addAll(createSplits(path, blockMap, start, end, align)); start = end; } return results; }  }
TemporaryInputFormat extends InputFormat<NullWritable, T> { static List<FileSplit> computeSplits(Path path, BlockMap blockMap, long splitSize) { long align = splitSize; if (splitSize > 0) { long remain = splitSize % TemporaryFile.BLOCK_SIZE; if (remain != 0) { align += TemporaryFile.BLOCK_SIZE - remain; } } long size = blockMap.getFileSize(); long start = 0; List<FileSplit> results = new ArrayList<>(); for (BlockInfo block : blockMap.getBlocks()) { assert start % TemporaryFile.BLOCK_SIZE == 0; long end = block.getEnd(); if (end < start) { continue; } long remain = end % TemporaryFile.BLOCK_SIZE; if (remain != 0) { end = Math.min(size, end + (TemporaryFile.BLOCK_SIZE - remain)); } results.addAll(createSplits(path, blockMap, start, end, align)); start = end; } return results; }  @Override List<InputSplit> getSplits(JobContext context); List<InputSplit> getSplits(
            JobContext context,
            List<StageInput> inputList); static void setInputPaths(Job job, List<Path> paths); static List<Path> getInputPaths(JobContext context); @Override RecordReader<NullWritable, T> createRecordReader(
            InputSplit split,
            TaskAttemptContext context); }
TemporaryInputFormat extends InputFormat<NullWritable, T> { static List<FileSplit> computeSplits(Path path, BlockMap blockMap, long splitSize) { long align = splitSize; if (splitSize > 0) { long remain = splitSize % TemporaryFile.BLOCK_SIZE; if (remain != 0) { align += TemporaryFile.BLOCK_SIZE - remain; } } long size = blockMap.getFileSize(); long start = 0; List<FileSplit> results = new ArrayList<>(); for (BlockInfo block : blockMap.getBlocks()) { assert start % TemporaryFile.BLOCK_SIZE == 0; long end = block.getEnd(); if (end < start) { continue; } long remain = end % TemporaryFile.BLOCK_SIZE; if (remain != 0) { end = Math.min(size, end + (TemporaryFile.BLOCK_SIZE - remain)); } results.addAll(createSplits(path, blockMap, start, end, align)); start = end; } return results; }  @Override List<InputSplit> getSplits(JobContext context); List<InputSplit> getSplits(
            JobContext context,
            List<StageInput> inputList); static void setInputPaths(Job job, List<Path> paths); static List<Path> getInputPaths(JobContext context); @Override RecordReader<NullWritable, T> createRecordReader(
            InputSplit split,
            TaskAttemptContext context);  }