DynamicInputFormat extends InputFormat<K, V> { static int getSplitRatio(int nMaps, int nRecords) { if (nMaps == 1) { LOG.warn("nMaps == 1. Why use DynamicInputFormat?"); return 1; } if (nMaps > MAX_CHUNKS_IDEAL) return SPLIT_RATIO_DEFAULT; int nPickups = (int)Math.ceil((float)MAX_CHUNKS_IDEAL/nMaps); int nRecordsPerChunk = (int)Math.ceil((float)nRecords/(nMaps*nPickups)); return nRecordsPerChunk < MIN_RECORDS_PER_CHUNK ? SPLIT_RATIO_DEFAULT : nPickups; }  @Override List<InputSplit> getSplits(JobContext jobContext); @Override RecordReader<K, V> createRecordReader(
          InputSplit inputSplit,
          TaskAttemptContext taskAttemptContext);  }
@Test public void testGetSplitRatio() throws Exception { Assert.assertEquals(1, DynamicInputFormat.getSplitRatio(1, 1000000000)); Assert.assertEquals(2, DynamicInputFormat.getSplitRatio(11000000, 10)); Assert.assertEquals(4, DynamicInputFormat.getSplitRatio(30, 700)); Assert.assertEquals(2, DynamicInputFormat.getSplitRatio(30, 200)); }