THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
See the Javadoc for HiveEndPoint for more information. Generally a user will establish a connection to the end point the destination info with HiveEndPoint object and then call calls newConnection to make a connection and get back a StreamingConnection object.
StreamingConnection
The StreamingConnection class is used to acquire batches of transactions. Once the connection has been provided by HiveEndPoint the application will generally enter a loop where it calls fetchTransactionBatch and writes a series of transactions. When closing down, the application should call close. See the Javadoc for more information.
...
TransactionBatch is used to write a series of transactions. For each transaction, the application calls beginNextTransaction, write, and then commit or abort as appropriate. See the Javadoc for details. All records in a single Transaction Batch will go to the same bucket. The API randomly picks a bucket for each new TxnBatch in order to spread the data among buckets.
Notes about the HiveConf Object
...
| Code Block | ||
|---|---|---|
| ||
///// Stream five records in two transactions /////
// Assumed HIVE table Schema:
create table alerts ( id int , msg string )
partitioned by (continent string, country string)
clustered by (id) into 5 buckets
stored as orc; // currently ORC is required for streaming
//------- MAIN THREAD ------- //
String dbName = "testing";
String tblName = "alerts";
ArrayList<String> partitionVals = new ArrayList<String>(2);
partitionVals.add("Asia");
partitionVals.add("India");
String serdeClass = "org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe";
HiveEndPoint hiveEP = new HiveEndPoint("thrift://x.y.com:9083", dbName, tblName, partitionVals);
.. spin up threads ..
//------- Thread 1 -------//
StreamingConnection connection = hiveEP.newConnection(null, true);
DelimitedInputWriter writer =
new DelimitedInputWriter(fieldNames,",", endPt);
TransactionBatch txnBatch = connection.fetchTransactionBatch(10, writer);
///// Batch 1 - First TXN
txnBatch.beginNextTransaction();
txnBatch.write("1,Hello streaming".getBytes());
txnBatch.write("2,Welcome to streaming".getBytes());
txnBatch.commit();
if(txnBatch.remainingTransactions() > 0) {
///// Batch 1 - Second TXN
txnBatch.beginNextTransaction();
txnBatch.write("3,Roshan Naik".getBytes());
txnBatch.write("4,Alan Gates".getBytes());
txnBatch.write("5,Owen O’Malley".getBytes());
txnBatch.commit();
txnBatch.close();
connection.close();
}
txnBatch = connection.fetchTransactionBatch(10, writer);
///// Batch 2 - First TXN
txnBatch.beginNextTransaction();
txnBatch.write("6,David Schorow".getBytes());
txnBatch.write("7,Sushant Sowmyan".getBytes());
txnBatch.commit();
if(txnBatch.remainingTransactions() > 0) {
///// Batch 2 - Second TXN
txnBatch.beginNextTransaction();
txnBatch.write("8,Ashutosh Chauhan".getBytes());
txnBatch.write("9,Thejas Nair" getBytes());
txnBatch.commit();
txnBatch.close();
}
connection.close();
//------- Thread 2 -------//
StreamingConnection connection2 = hiveEP.newConnection(null, true);
DelimitedInputWriter writer2 =
new DelimitedInputWriter(fieldNames,",", endPt);
TransactionBatch txnBatch2= connection.fetchTransactionBatch(10, writer2);
///// Batch 1 - First TXN
txnBatch2.beginNextTransaction();
txnBatch2.write("21,Venkat Ranganathan".getBytes());
txnBatch2.write("22,Bowen Zhang".getBytes());
txnBatch2.commit();
///// Batch 1 - Second TXN
txnBatch2.beginNextTransaction();
txnBatch2.write("23,Venkatesh Seetaram".getBytes());
txnBatch2.write("24,Deepesh Khandelwal".getBytes());
txnBatch2.commit();
txnBatch2.close();
connection.close();
txnBatch = connection.fetchTransactionBatch(10, writer);
///// Batch 2 - First TXN
txnBatch.beginNextTransaction();
txnBatch.write("26,David Schorow".getBytes());
txnBatch.write("27,Sushant Sowmyan".getBytes());
txnBatch.commit();
txnBatch2.close();
connection2.close(); |
...