JDBCInputFormat extends RichInputFormat<OUT, InputSplit> implements NonParallelInput { public JDBCInputFormat() { } JDBCInputFormat(); @Override void configure(Configuration parameters); @Override void open(InputSplit ignored); @Override void close(); @Override boolean reachedEnd(); @Override OUT nextRecord(OUT tuple); @Override BaseStatistics getStatistics(BaseStatistics cachedStatistics); @Override InputSplit[] createInputSplits(int minNumSplits); @Override InputSplitAssigner getInputSplitAssigner(InputSplit[] inputSplits); static JDBCInputFormatBuilder buildJDBCInputFormat();  }
@Test public void testJDBCInputFormat() throws IOException { jdbcInputFormat = JDBCInputFormat.buildJDBCInputFormat() .setDrivername("org.apache.derby.jdbc.EmbeddedDriver") .setDBUrl("jdbc:derby:memory:ebookshop") .setQuery("select * from books") .setResultSetType(ResultSet.TYPE_SCROLL_INSENSITIVE) .finish(); jdbcInputFormat.open(null); Tuple5 tuple = new Tuple5(); int recordCount = 0; while (!jdbcInputFormat.reachedEnd()) { jdbcInputFormat.nextRecord(tuple); Assert.assertEquals("Field 0 should be int", Integer.class, tuple.getField(0).getClass()); Assert.assertEquals("Field 1 should be String", String.class, tuple.getField(1).getClass()); Assert.assertEquals("Field 2 should be String", String.class, tuple.getField(2).getClass()); Assert.assertEquals("Field 3 should be float", Double.class, tuple.getField(3).getClass()); Assert.assertEquals("Field 4 should be int", Integer.class, tuple.getField(4).getClass()); for (int x = 0; x < 5; x++) { Assert.assertEquals(dbData[recordCount][x], tuple.getField(x)); } recordCount++; } Assert.assertEquals(5, recordCount); }