@Test public void testGetSplits() throws Exception { URI baseUri = new URI(GobblinWorkUnitsInputFormatTest.class.getSimpleName() + ": Configuration configuration = new Configuration(); Path workUnitsDir = new Path(new Path(baseUri), "/workUnits"); FileSystem fs = Mockito.mock(FileSystem.class); FileStatus[] statuses = createFileStatuses(20, workUnitsDir); Mockito.when(fs.listStatus(workUnitsDir)).thenReturn(statuses); Mockito.when(fs.makeQualified(Mockito.any(Path.class))).thenAnswer(new Answer<Path>() { @Override public Path answer(InvocationOnMock invocation) throws Throwable { return (Path) invocation.getArguments()[0]; } }); FileSystemTestUtils.addFileSystemForTest(baseUri, configuration, fs); GobblinWorkUnitsInputFormat inputFormat = new GobblinWorkUnitsInputFormat(); Job job = Job.getInstance(configuration); FileInputFormat.addInputPath(job, workUnitsDir); List<InputSplit> splits = inputFormat.getSplits(job); Assert.assertEquals(splits.size(), 20); verifyPaths(splits, statuses); }
@Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Path[] inputPaths = FileInputFormat.getInputPaths(context); if (inputPaths == null || inputPaths.length == 0) { throw new IOException("No input found!"); } List<String> allPaths = Lists.newArrayList(); for (Path path : inputPaths) { FileSystem fs = path.getFileSystem(context.getConfiguration()); FileStatus[] inputs = fs.listStatus(path); if (inputs == null) { throw new IOException(String.format("Path %s does not exist.", path)); } log.info(String.format("Found %d input files at %s: %s", inputs.length, path, Arrays.toString(inputs))); for (FileStatus input : inputs) { allPaths.add(input.getPath().toString()); } } int maxMappers = getMaxMapper(context.getConfiguration()); int numTasksPerMapper = allPaths.size() % maxMappers == 0 ? allPaths.size() / maxMappers : allPaths.size() / maxMappers + 1; List<InputSplit> splits = Lists.newArrayList(); Iterator<String> pathsIt = allPaths.iterator(); while (pathsIt.hasNext()) { Iterator<String> limitedIterator = Iterators.limit(pathsIt, numTasksPerMapper); splits.add(new GobblinSplit(Lists.newArrayList(limitedIterator))); } return splits; }
GobblinWorkUnitsInputFormat extends InputFormat<LongWritable, Text> { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Path[] inputPaths = FileInputFormat.getInputPaths(context); if (inputPaths == null || inputPaths.length == 0) { throw new IOException("No input found!"); } List<String> allPaths = Lists.newArrayList(); for (Path path : inputPaths) { FileSystem fs = path.getFileSystem(context.getConfiguration()); FileStatus[] inputs = fs.listStatus(path); if (inputs == null) { throw new IOException(String.format("Path %s does not exist.", path)); } log.info(String.format("Found %d input files at %s: %s", inputs.length, path, Arrays.toString(inputs))); for (FileStatus input : inputs) { allPaths.add(input.getPath().toString()); } } int maxMappers = getMaxMapper(context.getConfiguration()); int numTasksPerMapper = allPaths.size() % maxMappers == 0 ? allPaths.size() / maxMappers : allPaths.size() / maxMappers + 1; List<InputSplit> splits = Lists.newArrayList(); Iterator<String> pathsIt = allPaths.iterator(); while (pathsIt.hasNext()) { Iterator<String> limitedIterator = Iterators.limit(pathsIt, numTasksPerMapper); splits.add(new GobblinSplit(Lists.newArrayList(limitedIterator))); } return splits; } }
GobblinWorkUnitsInputFormat extends InputFormat<LongWritable, Text> { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Path[] inputPaths = FileInputFormat.getInputPaths(context); if (inputPaths == null || inputPaths.length == 0) { throw new IOException("No input found!"); } List<String> allPaths = Lists.newArrayList(); for (Path path : inputPaths) { FileSystem fs = path.getFileSystem(context.getConfiguration()); FileStatus[] inputs = fs.listStatus(path); if (inputs == null) { throw new IOException(String.format("Path %s does not exist.", path)); } log.info(String.format("Found %d input files at %s: %s", inputs.length, path, Arrays.toString(inputs))); for (FileStatus input : inputs) { allPaths.add(input.getPath().toString()); } } int maxMappers = getMaxMapper(context.getConfiguration()); int numTasksPerMapper = allPaths.size() % maxMappers == 0 ? allPaths.size() / maxMappers : allPaths.size() / maxMappers + 1; List<InputSplit> splits = Lists.newArrayList(); Iterator<String> pathsIt = allPaths.iterator(); while (pathsIt.hasNext()) { Iterator<String> limitedIterator = Iterators.limit(pathsIt, numTasksPerMapper); splits.add(new GobblinSplit(Lists.newArrayList(limitedIterator))); } return splits; }  }
GobblinWorkUnitsInputFormat extends InputFormat<LongWritable, Text> { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Path[] inputPaths = FileInputFormat.getInputPaths(context); if (inputPaths == null || inputPaths.length == 0) { throw new IOException("No input found!"); } List<String> allPaths = Lists.newArrayList(); for (Path path : inputPaths) { FileSystem fs = path.getFileSystem(context.getConfiguration()); FileStatus[] inputs = fs.listStatus(path); if (inputs == null) { throw new IOException(String.format("Path %s does not exist.", path)); } log.info(String.format("Found %d input files at %s: %s", inputs.length, path, Arrays.toString(inputs))); for (FileStatus input : inputs) { allPaths.add(input.getPath().toString()); } } int maxMappers = getMaxMapper(context.getConfiguration()); int numTasksPerMapper = allPaths.size() % maxMappers == 0 ? allPaths.size() / maxMappers : allPaths.size() / maxMappers + 1; List<InputSplit> splits = Lists.newArrayList(); Iterator<String> pathsIt = allPaths.iterator(); while (pathsIt.hasNext()) { Iterator<String> limitedIterator = Iterators.limit(pathsIt, numTasksPerMapper); splits.add(new GobblinSplit(Lists.newArrayList(limitedIterator))); } return splits; }  static void setMaxMappers(Job job, int maxMappers); static int getMaxMapper(Configuration conf); @Override List<InputSplit> getSplits(JobContext context); @Override RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context); }
GobblinWorkUnitsInputFormat extends InputFormat<LongWritable, Text> { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Path[] inputPaths = FileInputFormat.getInputPaths(context); if (inputPaths == null || inputPaths.length == 0) { throw new IOException("No input found!"); } List<String> allPaths = Lists.newArrayList(); for (Path path : inputPaths) { FileSystem fs = path.getFileSystem(context.getConfiguration()); FileStatus[] inputs = fs.listStatus(path); if (inputs == null) { throw new IOException(String.format("Path %s does not exist.", path)); } log.info(String.format("Found %d input files at %s: %s", inputs.length, path, Arrays.toString(inputs))); for (FileStatus input : inputs) { allPaths.add(input.getPath().toString()); } } int maxMappers = getMaxMapper(context.getConfiguration()); int numTasksPerMapper = allPaths.size() % maxMappers == 0 ? allPaths.size() / maxMappers : allPaths.size() / maxMappers + 1; List<InputSplit> splits = Lists.newArrayList(); Iterator<String> pathsIt = allPaths.iterator(); while (pathsIt.hasNext()) { Iterator<String> limitedIterator = Iterators.limit(pathsIt, numTasksPerMapper); splits.add(new GobblinSplit(Lists.newArrayList(limitedIterator))); } return splits; }  static void setMaxMappers(Job job, int maxMappers); static int getMaxMapper(Configuration conf); @Override List<InputSplit> getSplits(JobContext context); @Override RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context);  }