@Test public void testRecordReader() throws Exception { int n = 1; int maxNumSplits = 100; int maxNbTrees = 1000; Random rng = RandomUtils.getRandom(); for (int nloop = 0; nloop < n; nloop++) { int numSplits = rng.nextInt(maxNumSplits) + 1; int nbTrees = rng.nextInt(maxNbTrees) + 1; Configuration conf = getConfiguration(); Builder.setNbTrees(conf, nbTrees); InMemInputFormat inputFormat = new InMemInputFormat(); List<InputSplit> splits = inputFormat.getSplits(conf, numSplits); for (int index = 0; index < numSplits; index++) { InMemInputSplit split = (InMemInputSplit) splits.get(index); InMemRecordReader reader = new InMemRecordReader(split); reader.initialize(split, null); for (int tree = 0; tree < split.getNbTrees(); tree++) { assertEquals(tree < split.getNbTrees(), reader.nextKeyValue()); assertEquals(split.getFirstId() + tree, reader.getCurrentKey().get()); } } } }
@Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Configuration conf = context.getConfiguration(); int numSplits = conf.getInt("mapred.map.tasks", -1); return getSplits(conf, numSplits); }
InMemInputFormat extends InputFormat<IntWritable,NullWritable> { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Configuration conf = context.getConfiguration(); int numSplits = conf.getInt("mapred.map.tasks", -1); return getSplits(conf, numSplits); } }
InMemInputFormat extends InputFormat<IntWritable,NullWritable> { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Configuration conf = context.getConfiguration(); int numSplits = conf.getInt("mapred.map.tasks", -1); return getSplits(conf, numSplits); }  }
InMemInputFormat extends InputFormat<IntWritable,NullWritable> { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Configuration conf = context.getConfiguration(); int numSplits = conf.getInt("mapred.map.tasks", -1); return getSplits(conf, numSplits); }  @Override RecordReader<IntWritable,NullWritable> createRecordReader(InputSplit split, TaskAttemptContext context); @Override List<InputSplit> getSplits(JobContext context); List<InputSplit> getSplits(Configuration conf, int numSplits); }
InMemInputFormat extends InputFormat<IntWritable,NullWritable> { @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException { Configuration conf = context.getConfiguration(); int numSplits = conf.getInt("mapred.map.tasks", -1); return getSplits(conf, numSplits); }  @Override RecordReader<IntWritable,NullWritable> createRecordReader(InputSplit split, TaskAttemptContext context); @Override List<InputSplit> getSplits(JobContext context); List<InputSplit> getSplits(Configuration conf, int numSplits);  }