NIFI-626: Code cleanup to adhere to NiFi coding styles

This commit is contained in:
Mark Payne 2015-06-10 12:16:52 -04:00
parent d98757e4c7
commit a3b8e44ad5
5 changed files with 266 additions and 279 deletions

View File

@ -44,6 +44,7 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback; import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators; import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processors.standard.util.JdbcCommon; import org.apache.nifi.processors.standard.util.JdbcCommon;
import org.apache.nifi.util.LongHolder;
import org.apache.nifi.util.StopWatch; import org.apache.nifi.util.StopWatch;
@EventDriven @EventDriven
@ -91,11 +92,11 @@ public class ExecuteSQL extends AbstractProcessor {
private final List<PropertyDescriptor> propDescriptors; private final List<PropertyDescriptor> propDescriptors;
public ExecuteSQL() { public ExecuteSQL() {
HashSet<Relationship> r = new HashSet<>(); final Set<Relationship> r = new HashSet<>();
r.add(REL_SUCCESS); r.add(REL_SUCCESS);
relationships = Collections.unmodifiableSet(r); relationships = Collections.unmodifiableSet(r);
ArrayList<PropertyDescriptor> pds = new ArrayList<>(); final List<PropertyDescriptor> pds = new ArrayList<>();
pds.add(DBCP_SERVICE); pds.add(DBCP_SERVICE);
pds.add(SQL_SELECT_QUERY); pds.add(SQL_SELECT_QUERY);
pds.add(QUERY_TIMEOUT); pds.add(QUERY_TIMEOUT);
@ -113,7 +114,7 @@ public class ExecuteSQL extends AbstractProcessor {
} }
@Override @Override
public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
FlowFile incoming = session.get(); FlowFile incoming = session.get();
if (incoming == null) { if (incoming == null) {
return; return;
@ -127,43 +128,29 @@ public class ExecuteSQL extends AbstractProcessor {
final StopWatch stopWatch = new StopWatch(true); final StopWatch stopWatch = new StopWatch(true);
try { try (final Connection con = dbcpService.getConnection();
final Connection con = dbcpService.getConnection(); final Statement st = con.createStatement()) {
try {
final Statement st = con.createStatement();
try {
st.setQueryTimeout(queryTimeout); // timeout in seconds st.setQueryTimeout(queryTimeout); // timeout in seconds
final LongHolder nrOfRows = new LongHolder(0L);
FlowFile outgoing = session.write(incoming, new OutputStreamCallback() { FlowFile outgoing = session.write(incoming, new OutputStreamCallback() {
@Override @Override
public void process(final OutputStream out) throws IOException { public void process(final OutputStream out) throws IOException {
try { try {
logger.info("start executing query {}", new Object[]{selectQuery}); logger.debug("Executing query {}", new Object[] { selectQuery });
ResultSet resultSet = st.executeQuery(selectQuery); final ResultSet resultSet = st.executeQuery(selectQuery);
Long nrOfRows = JdbcCommon.convertToAvroStream(resultSet, out); nrOfRows.set(JdbcCommon.convertToAvroStream(resultSet, out));
logger.info("Result FlowFile contain {} Avro records", new Object[]{nrOfRows}); } catch (final SQLException e) {
} catch (SQLException e) {
throw new ProcessException(e); throw new ProcessException(e);
} }
} }
}); });
logger.info("Transferred {} to 'success'", new Object[]{outgoing}); logger.info("{} contains {} Avro records", new Object[] { nrOfRows.get() });
session.getProvenanceReporter().modifyContent(outgoing, stopWatch.getElapsed(TimeUnit.MILLISECONDS)); logger.info("Transferred {} to 'success'", new Object[] { outgoing });
session.getProvenanceReporter().modifyContent(outgoing, "Retrieved " + nrOfRows.get() + " rows", stopWatch.getElapsed(TimeUnit.MILLISECONDS));
session.transfer(outgoing, REL_SUCCESS); session.transfer(outgoing, REL_SUCCESS);
} finally { } catch (final ProcessException | SQLException e) {
st.close(); logger.error("Unable to execute SQL select query {} for {} due to {}; routing to failure", new Object[] { selectQuery, incoming, e });
}
} finally {
// return connection to pool
con.close();
}
} catch (ProcessException e) {
logger.error("Unable to execute sql select query due to {}", new Object[]{e});
session.transfer(incoming, REL_FAILURE);
} catch (SQLException e) {
logger.error("Unable to execute sql select query due to {}", new Object[]{e});
session.transfer(incoming, REL_FAILURE); session.transfer(incoming, REL_FAILURE);
} }
} }

View File

@ -21,6 +21,7 @@ import java.io.OutputStream;
import java.sql.ResultSet; import java.sql.ResultSet;
import java.sql.ResultSetMetaData; import java.sql.ResultSetMetaData;
import java.sql.SQLException; import java.sql.SQLException;
import static java.sql.Types.*;
import org.apache.avro.Schema; import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder; import org.apache.avro.SchemaBuilder;
@ -34,104 +35,102 @@ import org.apache.avro.io.DatumWriter;
/** /**
* JDBC / SQL common functions. * JDBC / SQL common functions.
*
*/ */
public class JdbcCommon { public class JdbcCommon {
public static long convertToAvroStream(ResultSet rs, OutputStream outStream) throws SQLException, IOException { public static long convertToAvroStream(final ResultSet rs, final OutputStream outStream) throws SQLException, IOException {
final Schema schema = createSchema(rs);
final GenericRecord rec = new GenericData.Record(schema);
Schema schema = createSchema(rs); final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema);
GenericRecord rec = new GenericData.Record(schema); try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<GenericRecord>(datumWriter)) {
DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<GenericRecord>(schema);
DataFileWriter<GenericRecord> dataFileWriter= new DataFileWriter<GenericRecord>(datumWriter);
dataFileWriter.create(schema, outStream); dataFileWriter.create(schema, outStream);
ResultSetMetaData meta = rs.getMetaData(); final ResultSetMetaData meta = rs.getMetaData();
int nrOfColumns = meta.getColumnCount(); final int nrOfColumns = meta.getColumnCount();
long nrOfRows = 0; long nrOfRows = 0;
while (rs.next()) { while (rs.next()) {
for (int i = 1; i <= nrOfColumns; i++) { for (int i = 1; i <= nrOfColumns; i++) {
Object value = rs.getObject(i); final Object value = rs.getObject(i);
rec.put(i-1, value); rec.put(i - 1, value);
} }
dataFileWriter.append(rec); dataFileWriter.append(rec);
nrOfRows += 1; nrOfRows += 1;
} }
dataFileWriter.close();
return nrOfRows; return nrOfRows;
} }
}
public static Schema createSchema(ResultSet rs) throws SQLException { public static Schema createSchema(final ResultSet rs) throws SQLException {
final ResultSetMetaData meta = rs.getMetaData();
final int nrOfColumns = meta.getColumnCount();
final String tableName = meta.getTableName(1);
ResultSetMetaData meta = rs.getMetaData(); final FieldAssembler<Schema> builder = SchemaBuilder.record(tableName).namespace("any.data").fields();
int nrOfColumns = meta.getColumnCount();
String tableName = meta.getTableName(1);
FieldAssembler<Schema> builder = SchemaBuilder.record(tableName).namespace("any.data").fields();
/** /**
* Some missing Avro types - Decimal, Date types. * Some missing Avro types - Decimal, Date types. May need some
* May need some additional work. * additional work.
*/ */
for (int i = 1; i <= nrOfColumns; i++) for (int i = 1; i <= nrOfColumns; i++) {
switch (meta.getColumnType(i)) { switch (meta.getColumnType(i)) {
case CHAR:
case java.sql.Types.CHAR: case LONGNVARCHAR:
case java.sql.Types.LONGNVARCHAR: case LONGVARCHAR:
case java.sql.Types.LONGVARCHAR: case NCHAR:
case java.sql.Types.NCHAR: case NVARCHAR:
case java.sql.Types.NVARCHAR: case VARCHAR:
case java.sql.Types.VARCHAR:
builder.name(meta.getColumnName(i)).type().stringType().noDefault(); builder.name(meta.getColumnName(i)).type().stringType().noDefault();
break; break;
case java.sql.Types.BOOLEAN: case BOOLEAN:
builder.name(meta.getColumnName(i)).type().booleanType().noDefault(); builder.name(meta.getColumnName(i)).type().booleanType().noDefault();
break; break;
case java.sql.Types.INTEGER: case INTEGER:
case java.sql.Types.SMALLINT: case SMALLINT:
case java.sql.Types.TINYINT: case TINYINT:
builder.name(meta.getColumnName(i)).type().intType().noDefault(); builder.name(meta.getColumnName(i)).type().intType().noDefault();
break; break;
case java.sql.Types.BIGINT: case BIGINT:
builder.name(meta.getColumnName(i)).type().longType().noDefault(); builder.name(meta.getColumnName(i)).type().longType().noDefault();
break; break;
// java.sql.RowId is interface, is seems to be database implementation specific, let's convert to String // java.sql.RowId is interface, is seems to be database
case java.sql.Types.ROWID: // implementation specific, let's convert to String
case ROWID:
builder.name(meta.getColumnName(i)).type().stringType().noDefault(); builder.name(meta.getColumnName(i)).type().stringType().noDefault();
break; break;
case java.sql.Types.FLOAT: case FLOAT:
case java.sql.Types.REAL: case REAL:
builder.name(meta.getColumnName(i)).type().floatType().noDefault(); builder.name(meta.getColumnName(i)).type().floatType().noDefault();
break; break;
case java.sql.Types.DOUBLE: case DOUBLE:
builder.name(meta.getColumnName(i)).type().doubleType().noDefault(); builder.name(meta.getColumnName(i)).type().doubleType().noDefault();
break; break;
// TODO Did not find direct suitable type, need to be clarified!!!! // Did not find direct suitable type, need to be clarified!!!!
case java.sql.Types.DECIMAL: case DECIMAL:
case java.sql.Types.NUMERIC: case NUMERIC:
builder.name(meta.getColumnName(i)).type().stringType().noDefault(); builder.name(meta.getColumnName(i)).type().stringType().noDefault();
break; break;
// TODO Did not find direct suitable type, need to be clarified!!!! // Did not find direct suitable type, need to be clarified!!!!
case java.sql.Types.DATE: case DATE:
case java.sql.Types.TIME: case TIME:
case java.sql.Types.TIMESTAMP: case TIMESTAMP:
builder.name(meta.getColumnName(i)).type().stringType().noDefault(); builder.name(meta.getColumnName(i)).type().stringType().noDefault();
break; break;
default: default:
break; break;
} }
return builder.endRecord();
} }
return builder.endRecord();
}
} }

View File

@ -77,7 +77,7 @@ public class TestExecuteSQL {
invokeOnTrigger(1); // 1 second max time invokeOnTrigger(1); // 1 second max time
} }
public void invokeOnTrigger(Integer queryTimeout) throws InitializationException, ClassNotFoundException, SQLException, IOException { public void invokeOnTrigger(final Integer queryTimeout) throws InitializationException, ClassNotFoundException, SQLException, IOException {
final TestRunner runner = TestRunners.newTestRunner(ExecuteSQL.class); final TestRunner runner = TestRunners.newTestRunner(ExecuteSQL.class);
final DBCPService dbcp = new DBCPServiceSimpleImpl(); final DBCPService dbcp = new DBCPServiceSimpleImpl();
@ -88,22 +88,23 @@ public class TestExecuteSQL {
runner.enableControllerService(dbcp); runner.enableControllerService(dbcp);
runner.setProperty(ExecuteSQL.DBCP_SERVICE, "dbcp"); runner.setProperty(ExecuteSQL.DBCP_SERVICE, "dbcp");
if (queryTimeout!=null) if (queryTimeout != null) {
runner.setProperty(ExecuteSQL.QUERY_TIMEOUT, queryTimeout.toString() + " secs"); runner.setProperty(ExecuteSQL.QUERY_TIMEOUT, queryTimeout.toString() + " secs");
}
// remove previous test database, if any // remove previous test database, if any
File dbLocation = new File(DB_LOCATION); final File dbLocation = new File(DB_LOCATION);
dbLocation.delete(); dbLocation.delete();
// load test data to database // load test data to database
Connection con = dbcp.getConnection(); final Connection con = dbcp.getConnection();
TestJdbcHugeStream.loadTestData2Database(con, 100, 2000, 1000); TestJdbcHugeStream.loadTestData2Database(con, 100, 2000, 1000);
System.out.println("test data loaded"); LOGGER.info("test data loaded");
// ResultSet size will be 1x2000x1000 = 2 000 000 rows // ResultSet size will be 1x2000x1000 = 2 000 000 rows
// because of where PER.ID = ${person.id} // because of where PER.ID = ${person.id}
final int nrOfRows = 2000000; final int nrOfRows = 2000000;
String query = "select " final String query = "select "
+ " PER.ID as PersonId, PER.NAME as PersonName, PER.CODE as PersonCode" + " PER.ID as PersonId, PER.NAME as PersonName, PER.CODE as PersonCode"
+ ", PRD.ID as ProductId,PRD.NAME as ProductName,PRD.CODE as ProductCode" + ", PRD.ID as ProductId,PRD.NAME as ProductName,PRD.CODE as ProductCode"
+ ", REL.ID as RelId, REL.NAME as RelName, REL.CODE as RelCode" + ", REL.ID as RelId, REL.NAME as RelName, REL.CODE as RelCode"
@ -114,7 +115,7 @@ public class TestExecuteSQL {
runner.setProperty(ExecuteSQL.SQL_SELECT_QUERY, query); runner.setProperty(ExecuteSQL.SQL_SELECT_QUERY, query);
// incoming FlowFile content is not used, but attributes are used // incoming FlowFile content is not used, but attributes are used
Map<String,String> attributes = new HashMap<String,String>(); final Map<String,String> attributes = new HashMap<String,String>();
attributes.put("person.id", "10"); attributes.put("person.id", "10");
runner.enqueue("Hello".getBytes(), attributes); runner.enqueue("Hello".getBytes(), attributes);
@ -122,20 +123,20 @@ public class TestExecuteSQL {
runner.assertAllFlowFilesTransferred(ExecuteSQL.REL_SUCCESS, 1); runner.assertAllFlowFilesTransferred(ExecuteSQL.REL_SUCCESS, 1);
// read all Avro records and verify created FlowFile contains 1000000 records // read all Avro records and verify created FlowFile contains 1000000 records
List<MockFlowFile> flowfiles = runner.getFlowFilesForRelationship(ExecuteSQL.REL_SUCCESS); final List<MockFlowFile> flowfiles = runner.getFlowFilesForRelationship(ExecuteSQL.REL_SUCCESS);
InputStream in = new ByteArrayInputStream(flowfiles.get(0).toByteArray()); final InputStream in = new ByteArrayInputStream(flowfiles.get(0).toByteArray());
DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(); final DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>();
DataFileStream<GenericRecord> dataFileReader = new DataFileStream<GenericRecord>(in, datumReader); final DataFileStream<GenericRecord> dataFileReader = new DataFileStream<GenericRecord>(in, datumReader);
GenericRecord record = null; GenericRecord record = null;
long recordsFromStream = 0; long recordsFromStream = 0;
while (dataFileReader.hasNext()) { while (dataFileReader.hasNext()) {
// Reuse record object by passing it to next(). This saves us from // Reuse record object by passing it to next(). This saves us from
// allocating and garbage collecting many objects for files with many items. // allocating and garbage collecting many objects for files with many items.
record = dataFileReader.next(record); record = dataFileReader.next(record);
// System.out.println(record);
recordsFromStream += 1; recordsFromStream += 1;
} }
System.out.println("total nr of records from stream: " + recordsFromStream);
LOGGER.info("total nr of records from stream: " + recordsFromStream);
assertEquals(nrOfRows, recordsFromStream); assertEquals(nrOfRows, recordsFromStream);
dataFileReader.close(); dataFileReader.close();
} }
@ -158,9 +159,9 @@ public class TestExecuteSQL {
public Connection getConnection() throws ProcessException { public Connection getConnection() throws ProcessException {
try { try {
Class.forName("org.apache.derby.jdbc.EmbeddedDriver"); Class.forName("org.apache.derby.jdbc.EmbeddedDriver");
Connection con = DriverManager.getConnection("jdbc:derby:" + DB_LOCATION + ";create=true"); final Connection con = DriverManager.getConnection("jdbc:derby:" + DB_LOCATION + ";create=true");
return con; return con;
} catch (Exception e) { } catch (final Exception e) {
throw new ProcessException("getConnection failed: " + e); throw new ProcessException("getConnection failed: " + e);
} }
} }

View File

@ -39,6 +39,7 @@ import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord; import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader; import org.apache.avro.io.DatumReader;
import org.junit.BeforeClass; import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test; import org.junit.Test;
/** /**
@ -73,118 +74,116 @@ public class TestJdbcHugeStream {
* at org.apache.avro.file.DataFileWriter$BufferedFileOutputStream$PositionFilter.write(DataFileWriter.java:446) * at org.apache.avro.file.DataFileWriter$BufferedFileOutputStream$PositionFilter.write(DataFileWriter.java:446)
* *
*/ */
// @Test @Test
@Ignore
public void readSend2StreamHuge_InMemory() throws ClassNotFoundException, SQLException, IOException { public void readSend2StreamHuge_InMemory() throws ClassNotFoundException, SQLException, IOException {
// remove previous test database, if any // remove previous test database, if any
File dbLocation = new File(DB_LOCATION); final File dbLocation = new File(DB_LOCATION);
dbLocation.delete(); dbLocation.delete();
Connection con = createConnection(); try (final Connection con = createConnection()) {
loadTestData2Database(con, 150, 150, 150); loadTestData2Database(con, 150, 150, 150);
System.out.println("test data loaded"); System.out.println("test data loaded");
Statement st = con.createStatement(); try (final Statement st = con.createStatement()) {
// Notice! // Notice!
// Following select is deliberately invalid! // Following select is deliberately invalid!
// For testing we need huge amount of rows, so where part is not used. // For testing we need huge amount of rows, so where part is not
ResultSet resultSet = st.executeQuery("select " // used.
final ResultSet resultSet = st.executeQuery("select "
+ " PER.ID as PersonId, PER.NAME as PersonName, PER.CODE as PersonCode" + " PER.ID as PersonId, PER.NAME as PersonName, PER.CODE as PersonCode"
+ ", PRD.ID as ProductId,PRD.NAME as ProductName,PRD.CODE as ProductCode" + ", PRD.ID as ProductId,PRD.NAME as ProductName,PRD.CODE as ProductCode"
+ ", REL.ID as RelId, REL.NAME as RelName, REL.CODE as RelCode" + ", REL.ID as RelId, REL.NAME as RelName, REL.CODE as RelCode"
+ ", ROW_NUMBER() OVER () as rownr " + ", ROW_NUMBER() OVER () as rownr "
+ " from persons PER, products PRD, relationships REL"); + " from persons PER, products PRD, relationships REL");
ByteArrayOutputStream outStream = new ByteArrayOutputStream(); final ByteArrayOutputStream outStream = new ByteArrayOutputStream();
long nrOfRows = JdbcCommon.convertToAvroStream(resultSet, outStream); final long nrOfRows = JdbcCommon.convertToAvroStream(resultSet, outStream);
System.out.println("total nr of rows in resultset: " + nrOfRows); System.out.println("total nr of rows in resultset: " + nrOfRows);
byte[] serializedBytes = outStream.toByteArray(); final byte[] serializedBytes = outStream.toByteArray();
assertNotNull(serializedBytes); assertNotNull(serializedBytes);
System.out.println("Avro serialized result size in bytes: " + serializedBytes.length); System.out.println("Avro serialized result size in bytes: " + serializedBytes.length);
// Deserialize bytes to records // Deserialize bytes to records
InputStream instream = new ByteArrayInputStream(serializedBytes); final InputStream instream = new ByteArrayInputStream(serializedBytes);
DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(); final DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>();
DataFileStream<GenericRecord> dataFileReader = new DataFileStream<GenericRecord>(instream, datumReader); try (final DataFileStream<GenericRecord> dataFileReader = new DataFileStream<GenericRecord>(instream, datumReader)) {
GenericRecord record = null; GenericRecord record = null;
long recordsFromStream = 0; long recordsFromStream = 0;
while (dataFileReader.hasNext()) { while (dataFileReader.hasNext()) {
// Reuse record object by passing it to next(). This saves us from // Reuse record object by passing it to next(). This
// allocating and garbage collecting many objects for files with many items. // saves us from
// allocating and garbage collecting many objects for
// files with many items.
record = dataFileReader.next(record); record = dataFileReader.next(record);
// System.out.println(record);
recordsFromStream += 1; recordsFromStream += 1;
} }
System.out.println("total nr of records from stream: " + recordsFromStream); System.out.println("total nr of records from stream: " + recordsFromStream);
assertEquals(nrOfRows, recordsFromStream); assertEquals(nrOfRows, recordsFromStream);
st.close(); }
con.close(); }
}
} }
@Test @Test
public void readSend2StreamHuge_FileBased() throws ClassNotFoundException, SQLException, IOException { public void readSend2StreamHuge_FileBased() throws ClassNotFoundException, SQLException, IOException {
// remove previous test database, if any // remove previous test database, if any
File dbLocation = new File(DB_LOCATION); final File dbLocation = new File(DB_LOCATION);
dbLocation.delete(); dbLocation.delete();
Connection con = createConnection(); try (final Connection con = createConnection()) {
loadTestData2Database(con, 300, 300, 300); loadTestData2Database(con, 300, 300, 300);
System.out.println("test data loaded");
Statement st = con.createStatement();
try (final Statement st = con.createStatement()) {
// Notice! // Notice!
// Following select is deliberately invalid! // Following select is deliberately invalid!
// For testing we need huge amount of rows, so where part is not used. // For testing we need huge amount of rows, so where part is not
ResultSet resultSet = st.executeQuery("select " // used.
final ResultSet resultSet = st.executeQuery("select "
+ " PER.ID as PersonId, PER.NAME as PersonName, PER.CODE as PersonCode" + " PER.ID as PersonId, PER.NAME as PersonName, PER.CODE as PersonCode"
+ ", PRD.ID as ProductId,PRD.NAME as ProductName,PRD.CODE as ProductCode" + ", PRD.ID as ProductId,PRD.NAME as ProductName,PRD.CODE as ProductCode"
+ ", REL.ID as RelId, REL.NAME as RelName, REL.CODE as RelCode" + ", REL.ID as RelId, REL.NAME as RelName, REL.CODE as RelCode"
+ ", ROW_NUMBER() OVER () as rownr " + ", ROW_NUMBER() OVER () as rownr "
+ " from persons PER, products PRD, relationships REL"); + " from persons PER, products PRD, relationships REL");
OutputStream outStream = new FileOutputStream("target/data.avro"); final OutputStream outStream = new FileOutputStream("target/data.avro");
long nrOfRows = JdbcCommon.convertToAvroStream(resultSet, outStream); final long nrOfRows = JdbcCommon.convertToAvroStream(resultSet, outStream);
System.out.println("total nr of rows in resultset: " + nrOfRows);
/*
byte[] serializedBytes = outStream.toByteArray();
assertNotNull(serializedBytes);
System.out.println("Avro serialized result size in bytes: " + serializedBytes.length);
*/
// Deserialize bytes to records // Deserialize bytes to records
final InputStream instream = new FileInputStream("target/data.avro");
InputStream instream = new FileInputStream("target/data.avro"); final DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>();
try (final DataFileStream<GenericRecord> dataFileReader = new DataFileStream<GenericRecord>(instream, datumReader)) {
DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>();
DataFileStream<GenericRecord> dataFileReader = new DataFileStream<GenericRecord>(instream, datumReader);
GenericRecord record = null; GenericRecord record = null;
long recordsFromStream = 0; long recordsFromStream = 0;
while (dataFileReader.hasNext()) { while (dataFileReader.hasNext()) {
// Reuse record object by passing it to next(). This saves us from // Reuse record object by passing it to next(). This
// allocating and garbage collecting many objects for files with many items. // saves us from
// allocating and garbage collecting many objects for
// files with many items.
record = dataFileReader.next(record); record = dataFileReader.next(record);
// System.out.println(record);
recordsFromStream += 1; recordsFromStream += 1;
} }
System.out.println("total nr of records from stream: " + recordsFromStream); System.out.println("total nr of records from stream: " + recordsFromStream);
assertEquals(nrOfRows, recordsFromStream); assertEquals(nrOfRows, recordsFromStream);
st.close(); }
con.close(); }
}
} }
//================================================ helpers =============================================== //================================================ helpers ===============================================
static String dropPersons = "drop table persons"; static String dropPersons = "drop table persons";
static String dropProducts = "drop table products"; static String dropProducts = "drop table products";
static String dropRelationships= "drop table relationships"; static String dropRelationships = "drop table relationships";
static String createPersons = "create table persons (id integer, name varchar(100), code integer)"; static String createPersons = "create table persons (id integer, name varchar(100), code integer)";
static String createProducts = "create table products (id integer, name varchar(100), code integer)"; static String createProducts = "create table products (id integer, name varchar(100), code integer)";
static String createRelationships = "create table relationships(id integer,name varchar(100), code integer)"; static String createRelationships = "create table relationships (id integer,name varchar(100), code integer)";
static public void loadTestData2Database(Connection con, int nrOfPersons, int nrOfProducts, int nrOfRels) throws ClassNotFoundException, SQLException { static public void loadTestData2Database(Connection con, int nrOfPersons, int nrOfProducts, int nrOfRels) throws ClassNotFoundException, SQLException {
@ -192,17 +191,17 @@ public class TestJdbcHugeStream {
System.out.println(createRandomName()); System.out.println(createRandomName());
System.out.println(createRandomName()); System.out.println(createRandomName());
Statement st = con.createStatement(); final Statement st = con.createStatement();
// tables may not exist, this is not serious problem. // tables may not exist, this is not serious problem.
try { st.executeUpdate(dropPersons); try { st.executeUpdate(dropPersons);
} catch (Exception e) { } } catch (final Exception e) { }
try { st.executeUpdate(dropProducts); try { st.executeUpdate(dropProducts);
} catch (Exception e) { } } catch (final Exception e) { }
try { st.executeUpdate(dropRelationships); try { st.executeUpdate(dropRelationships);
} catch (Exception e) { } } catch (final Exception e) { }
st.executeUpdate(createPersons); st.executeUpdate(createPersons);
st.executeUpdate(createProducts); st.executeUpdate(createProducts);
@ -223,17 +222,14 @@ public class TestJdbcHugeStream {
static Random rng = new Random(53495); static Random rng = new Random(53495);
static private void loadPersons(Statement st, int nr) throws SQLException { static private void loadPersons(Statement st, int nr) throws SQLException {
st.executeUpdate("insert into persons values (" + nr + ", '" + createRandomName() + "', " + rng.nextInt(469946) + ")" ); st.executeUpdate("insert into persons values (" + nr + ", '" + createRandomName() + "', " + rng.nextInt(469946) + ")" );
} }
static private void loadProducts(Statement st, int nr) throws SQLException { static private void loadProducts(Statement st, int nr) throws SQLException {
st.executeUpdate("insert into products values (" + nr + ", '" + createRandomName() + "', " + rng.nextInt(469946) + ")" ); st.executeUpdate("insert into products values (" + nr + ", '" + createRandomName() + "', " + rng.nextInt(469946) + ")" );
} }
static private void loadRelationships(Statement st, int nr) throws SQLException { static private void loadRelationships(Statement st, int nr) throws SQLException {
st.executeUpdate("insert into relationships values (" + nr + ", '" + createRandomName() + "', " + rng.nextInt(469946) + ")" ); st.executeUpdate("insert into relationships values (" + nr + ", '" + createRandomName() + "', " + rng.nextInt(469946) + ")" );
} }
@ -243,10 +239,10 @@ public class TestJdbcHugeStream {
static private String createRandomString() { static private String createRandomString() {
int length = rng.nextInt(19); final int length = rng.nextInt(19);
String characters = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"; final String characters = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
char[] text = new char[length]; final char[] text = new char[length];
for (int i = 0; i < length; i++) for (int i = 0; i < length; i++)
{ {
text[i] = characters.charAt(rng.nextInt(characters.length())); text[i] = characters.charAt(rng.nextInt(characters.length()));
@ -255,9 +251,8 @@ public class TestJdbcHugeStream {
} }
private Connection createConnection() throws ClassNotFoundException, SQLException { private Connection createConnection() throws ClassNotFoundException, SQLException {
Class.forName("org.apache.derby.jdbc.EmbeddedDriver"); Class.forName("org.apache.derby.jdbc.EmbeddedDriver");
Connection con = DriverManager.getConnection("jdbc:derby:" + DB_LOCATION + ";create=true"); final Connection con = DriverManager.getConnection("jdbc:derby:" + DB_LOCATION + ";create=true");
return con; return con;
} }

View File

@ -857,6 +857,11 @@
<artifactId>json-path</artifactId> <artifactId>json-path</artifactId>
<version>2.0.0</version> <version>2.0.0</version>
</dependency> </dependency>
<dependency>
<groupId>org.apache.derby</groupId>
<artifactId>derby</artifactId>
<version>10.11.1.1</version>
</dependency>
</dependencies> </dependencies>
</dependencyManagement> </dependencyManagement>
<dependencies> <dependencies>