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);  }
@Test public void testSplits() 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); assertEquals(numSplits, splits.size()); int nbTreesPerSplit = nbTrees / numSplits; int totalTrees = 0; int expectedId = 0; for (int index = 0; index < numSplits; index++) { assertTrue(splits.get(index) instanceof InMemInputSplit); InMemInputSplit split = (InMemInputSplit) splits.get(index); assertEquals(expectedId, split.getFirstId()); if (index < numSplits - 1) { assertEquals(nbTreesPerSplit, split.getNbTrees()); } else { assertEquals(nbTrees - totalTrees, split.getNbTrees()); } totalTrees += split.getNbTrees(); expectedId += split.getNbTrees(); } } }
@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()); } } } }
@Test public void testSplits() 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 = new Configuration(); Builder.setNbTrees(conf, nbTrees); InMemInputFormat inputFormat = new InMemInputFormat(); List<InputSplit> splits = inputFormat.getSplits(conf, numSplits); assertEquals(numSplits, splits.size()); int nbTreesPerSplit = nbTrees / numSplits; int totalTrees = 0; int expectedId = 0; for (int index = 0; index < numSplits; index++) { assertTrue(splits.get(index) instanceof InMemInputSplit); InMemInputSplit split = (InMemInputSplit) splits.get(index); assertEquals(expectedId, split.getFirstId()); if (index < numSplits - 1) { assertEquals(nbTreesPerSplit, split.getNbTrees()); } else { assertEquals(nbTrees - totalTrees, split.getNbTrees()); } totalTrees += split.getNbTrees(); expectedId += split.getNbTrees(); } } }
@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 = new Configuration(); 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()); } } } }