JDBCOutputFormat extends RichOutputFormat<Row> { public JDBCOutputFormat() { } JDBCOutputFormat(); @Override void configure(Configuration parameters); @Override void open(int taskNumber, int numTasks); @Override void writeRecord(Row row); @Override void close(); static JDBCOutputFormatBuilder buildJDBCOutputFormat();  }
@Test public void testJDBCOutputFormat() throws IOException, SQLException { jdbcOutputFormat = JDBCOutputFormat.buildJDBCOutputFormat() .setDrivername(DRIVER_CLASS) .setDBUrl(DB_URL) .setQuery(String.format(INSERT_TEMPLATE, OUTPUT_TABLE)) .finish(); jdbcOutputFormat.open(0, 1); for (JDBCTestBase.TestEntry entry : TEST_DATA) { jdbcOutputFormat.writeRecord(toRow(entry)); } jdbcOutputFormat.close(); try ( Connection dbConn = DriverManager.getConnection(DB_URL); PreparedStatement statement = dbConn.prepareStatement(JDBCTestBase.SELECT_ALL_NEWBOOKS); ResultSet resultSet = statement.executeQuery() ) { int recordCount = 0; while (resultSet.next()) { assertEquals(TEST_DATA[recordCount].id, resultSet.getObject("id")); assertEquals(TEST_DATA[recordCount].title, resultSet.getObject("title")); assertEquals(TEST_DATA[recordCount].author, resultSet.getObject("author")); assertEquals(TEST_DATA[recordCount].price, resultSet.getObject("price")); assertEquals(TEST_DATA[recordCount].qty, resultSet.getObject("qty")); recordCount++; } assertEquals(TEST_DATA.length, recordCount); } }