You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by bhupeshchawda <gi...@git.apache.org> on 2016/03/17 10:22:56 UTC
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
GitHub user bhupeshchawda opened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/212
APEXMALHAR-1957: Added threading for reading data from hbase. Added s…
…upport for progressive read. Added unit tests.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/bhupeshchawda/incubator-apex-malhar MLHR-1957-hbase-input
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/212.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #212
----
commit c8e45f27b363efb0fa8f22687c3098953385d509
Author: bhupesh <bh...@gmail.com>
Date: 2015-12-24T13:20:25Z
APEXMALHAR-1957: Added threading for reading data from hbase. Added support for progressive read. Added unit tests.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61075471
--- Diff: contrib/src/test/java/com/datatorrent/contrib/hbase/HBasePOJOInputOperatorTest.java ---
@@ -128,7 +144,7 @@ public void populateDAG(DAG dag, Configuration conf)
Thread.sleep(1000);
}
catch( Exception e ){}
-
+ logger.info("Tuple row key: ", output.getReceivedTuples());
logger.info( "Received tuple number {}, instance is {}.", output.getReceivedTuples() == null ? 0 : output.getReceivedTuples().size(), System.identityHashCode( output ) );
if( output.getReceivedTuples() != null && output.getReceivedTuples().size() == TUPLE_NUM )
--- End diff --
If this condition is never met, the unit test case will run forever. Need some timeout based failure of the test case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61073466
--- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBasePOJOInputOperator.java ---
@@ -69,116 +69,103 @@ public Object convertValue( HBaseFieldInfo fieldInfo, Object value)
public void setup(OperatorContext context)
{
try {
- store.connect();
+ super.setup(context);
pojoType = Class.forName(pojoTypeName);
pojoType.newInstance(); //try create new instance to verify the class.
rowSetter = PojoUtils.createSetter(pojoType, tableInfo.getRowOrIdExpression(), String.class);
- fieldValueGenerator = FieldValueGenerator.getFieldValueGenerator(pojoType, tableInfo.getFieldsInfo() );
+ fieldValueGenerator = HBaseFieldValueGenerator.getHBaseFieldValueGenerator(pojoType, tableInfo.getFieldsInfo() );
valueConverter = new BytesValueConverter();
} catch (Exception ex) {
throw new RuntimeException(ex);
}
}
@Override
- public void beginWindow(long windowId)
- {
- }
-
- @Override
- public void teardown()
+ protected Object getTuple(Result result)
{
try {
- store.disconnect();
- } catch (IOException ex) {
- throw new RuntimeException(ex);
- }
- }
-
- @Override
- public void emitTuples()
- {
- try {
- Scan scan = nextScan();
- if (scan == null)
- return;
-
- ResultScanner resultScanner = store.getTable().getScanner(scan);
-
- while (true) {
- Result result = resultScanner.next();
- if (result == null)
- break;
-
- String readRow = Bytes.toString(result.getRow());
- if( readRow.equals( lastReadRow ))
- continue;
-
- Object instance = pojoType.newInstance();
- rowSetter.set(instance, readRow);
-
- List<Cell> cells = result.listCells();
+ String readRow = Bytes.toString(result.getRow());
+ if( readRow.equals( getLastReadRow() )) {
+ return null;
+ }
- for (Cell cell : cells) {
- String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
- byte[] value = CellUtil.cloneValue(cell);
- fieldValueGenerator.setColumnValue( instance, columnName, value, valueConverter );
- }
+ Object instance = pojoType.newInstance();
+ rowSetter.set(instance, readRow);
- outputPort.emit(instance);
- lastReadRow = readRow;
+ List<Cell> cells = result.listCells();
+ for (Cell cell : cells) {
+ String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
+ String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
+ byte[] value = CellUtil.cloneValue(cell);
+ ((HBaseFieldValueGenerator)fieldValueGenerator).setColumnValue(instance, columnName, columnFamily, value,
+ valueConverter);
}
+ setLastReadRow(readRow);
+ return instance;
} catch (Exception e) {
- throw new RuntimeException(e.getMessage());
+ throw new RuntimeException(e);
}
-
- }
-
- protected Scan nextScan()
- {
- if(lastReadRow==null && startRow==null )
- return new Scan();
- else
- return new Scan( Bytes.toBytes( lastReadRow == null ? startRow : lastReadRow ) );
}
- public HBaseStore getStore()
- {
- return store;
- }
- public void setStore(HBaseStore store)
+ @Override
+ protected Scan operationScan()
{
- this.store = store;
+ Scan scan;
+ if (getLastReadRow() == null && getStartRow() == null) {
+ // If no start row specified and no row read yet
+ scan = new Scan();
--- End diff --
new scan() can be moved to setup()
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by bhupeshchawda <gi...@git.apache.org>.
Github user bhupeshchawda commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61719921
--- Diff: contrib/src/test/java/com/datatorrent/contrib/hbase/HBasePOJOInputOperatorTest.java ---
@@ -128,7 +144,7 @@ public void populateDAG(DAG dag, Configuration conf)
Thread.sleep(1000);
}
catch( Exception e ){}
-
+ logger.info("Tuple row key: ", output.getReceivedTuples());
logger.info( "Received tuple number {}, instance is {}.", output.getReceivedTuples() == null ? 0 : output.getReceivedTuples().size(), System.identityHashCode( output ) );
if( output.getReceivedTuples() != null && output.getReceivedTuples().size() == TUPLE_NUM )
--- End diff --
done. Added a check for 30 seconds.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61074411
--- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBaseScanOperator.java ---
@@ -40,27 +51,106 @@
* @tags hbase, scan, input operator
* @since 0.3.2
*/
-public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T>
+public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T> implements Operator.ActivationListener<Context>
{
+ public static final int DEF_HINT_SCAN_LOOKAHEAD = 2;
+ public static final int DEF_QUEUE_SIZE = 1000;
+ public static final int DEF_SLEEP_MILLIS = 10;
+
+ private String startRow;
+ private String endRow;
+ private String lastReadRow;
+ private int hintScanLookahead = DEF_HINT_SCAN_LOOKAHEAD;
+ private int queueSize = DEF_QUEUE_SIZE;
+ private int sleepMillis = DEF_SLEEP_MILLIS;
+ private Queue<Result> resultQueue;
+
+ @AutoMetric
+ protected long tuplesRead;
+
+ // Transients
+ protected transient Scan scan;
+ protected transient ResultScanner scanner;
+ protected transient Thread readThread;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ resultQueue = Queues.newLinkedBlockingQueue(queueSize);
+ }
+
+ @Override
+ public void activate(Context context)
+ {
+ startReadThread();
+ }
+
+ protected void startReadThread()
+ {
+ try {
+ scan = operationScan();
+ scanner = getStore().getTable().getScanner(scan);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ readThread = new Thread(new Runnable() {
+ @Override
+ public void run()
+ {
+ try {
+ Result result;
+ while ((result = scanner.next()) != null) {
+ while (!resultQueue.offer(result)) {
+ Thread.sleep(sleepMillis);
+ }
+ }
+ } catch (Exception e) {
+ logger.debug("Exception in fetching results {}", e.getMessage());
+ throw new RuntimeException(e);
+ } finally {
+ scanner.close();
+ }
+ }
+ });
+ readThread.start();
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ tuplesRead = 0;
+ }
@Override
public void emitTuples()
{
+ if (!readThread.isAlive() && resultQueue.isEmpty()) {
+ startReadThread();
--- End diff --
Better to throw RuntimeException and redeploy the operator.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by bhupeshchawda <gi...@git.apache.org>.
Github user bhupeshchawda closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/212
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by bhupeshchawda <gi...@git.apache.org>.
Github user bhupeshchawda commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61719497
--- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBaseScanOperator.java ---
@@ -40,27 +51,106 @@
* @tags hbase, scan, input operator
* @since 0.3.2
*/
-public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T>
+public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T> implements Operator.ActivationListener<Context>
{
+ public static final int DEF_HINT_SCAN_LOOKAHEAD = 2;
+ public static final int DEF_QUEUE_SIZE = 1000;
+ public static final int DEF_SLEEP_MILLIS = 10;
+
+ private String startRow;
+ private String endRow;
+ private String lastReadRow;
+ private int hintScanLookahead = DEF_HINT_SCAN_LOOKAHEAD;
+ private int queueSize = DEF_QUEUE_SIZE;
+ private int sleepMillis = DEF_SLEEP_MILLIS;
+ private Queue<Result> resultQueue;
+
+ @AutoMetric
+ protected long tuplesRead;
+
+ // Transients
+ protected transient Scan scan;
+ protected transient ResultScanner scanner;
+ protected transient Thread readThread;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ resultQueue = Queues.newLinkedBlockingQueue(queueSize);
+ }
+
+ @Override
+ public void activate(Context context)
+ {
+ startReadThread();
+ }
+
+ protected void startReadThread()
+ {
+ try {
+ scan = operationScan();
+ scanner = getStore().getTable().getScanner(scan);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ readThread = new Thread(new Runnable() {
+ @Override
+ public void run()
+ {
+ try {
+ Result result;
+ while ((result = scanner.next()) != null) {
+ while (!resultQueue.offer(result)) {
+ Thread.sleep(sleepMillis);
+ }
+ }
+ } catch (Exception e) {
+ logger.debug("Exception in fetching results {}", e.getMessage());
+ throw new RuntimeException(e);
+ } finally {
+ scanner.close();
+ }
+ }
+ });
+ readThread.start();
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ tuplesRead = 0;
+ }
@Override
public void emitTuples()
{
+ if (!readThread.isAlive() && resultQueue.isEmpty()) {
+ startReadThread();
+ }
try {
- HTable table = getTable();
- Scan scan = operationScan();
- ResultScanner scanner = table.getScanner(scan);
- for (Result result : scanner) {
- //KeyValue[] kvs = result.raw();
- //T t = getTuple(kvs);
- T t = getTuple(result);
- outputPort.emit(t);
+ Result result = resultQueue.poll();
+ if (result == null) {
+ Thread.sleep(sleepMillis);
+ return;
+ }
+ T tuple = getTuple(result);
+ if (tuple != null) {
+ outputPort.emit(tuple);
+ tuplesRead++;
}
} catch (Exception e) {
e.printStackTrace();
--- End diff --
done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61072959
--- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBasePOJOInputOperator.java ---
@@ -69,116 +69,103 @@ public Object convertValue( HBaseFieldInfo fieldInfo, Object value)
public void setup(OperatorContext context)
{
try {
- store.connect();
+ super.setup(context);
pojoType = Class.forName(pojoTypeName);
pojoType.newInstance(); //try create new instance to verify the class.
rowSetter = PojoUtils.createSetter(pojoType, tableInfo.getRowOrIdExpression(), String.class);
- fieldValueGenerator = FieldValueGenerator.getFieldValueGenerator(pojoType, tableInfo.getFieldsInfo() );
+ fieldValueGenerator = HBaseFieldValueGenerator.getHBaseFieldValueGenerator(pojoType, tableInfo.getFieldsInfo() );
valueConverter = new BytesValueConverter();
} catch (Exception ex) {
throw new RuntimeException(ex);
}
}
@Override
- public void beginWindow(long windowId)
- {
- }
-
- @Override
- public void teardown()
+ protected Object getTuple(Result result)
{
try {
- store.disconnect();
- } catch (IOException ex) {
- throw new RuntimeException(ex);
- }
- }
-
- @Override
- public void emitTuples()
- {
- try {
- Scan scan = nextScan();
- if (scan == null)
- return;
-
- ResultScanner resultScanner = store.getTable().getScanner(scan);
-
- while (true) {
- Result result = resultScanner.next();
- if (result == null)
- break;
-
- String readRow = Bytes.toString(result.getRow());
- if( readRow.equals( lastReadRow ))
- continue;
-
- Object instance = pojoType.newInstance();
- rowSetter.set(instance, readRow);
-
- List<Cell> cells = result.listCells();
+ String readRow = Bytes.toString(result.getRow());
+ if( readRow.equals( getLastReadRow() )) {
+ return null;
+ }
- for (Cell cell : cells) {
- String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
- byte[] value = CellUtil.cloneValue(cell);
- fieldValueGenerator.setColumnValue( instance, columnName, value, valueConverter );
- }
+ Object instance = pojoType.newInstance();
+ rowSetter.set(instance, readRow);
- outputPort.emit(instance);
- lastReadRow = readRow;
+ List<Cell> cells = result.listCells();
+ for (Cell cell : cells) {
+ String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
+ String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
+ byte[] value = CellUtil.cloneValue(cell);
+ ((HBaseFieldValueGenerator)fieldValueGenerator).setColumnValue(instance, columnName, columnFamily, value,
+ valueConverter);
}
+ setLastReadRow(readRow);
+ return instance;
} catch (Exception e) {
- throw new RuntimeException(e.getMessage());
+ throw new RuntimeException(e);
}
-
- }
-
- protected Scan nextScan()
- {
- if(lastReadRow==null && startRow==null )
- return new Scan();
- else
- return new Scan( Bytes.toBytes( lastReadRow == null ? startRow : lastReadRow ) );
}
- public HBaseStore getStore()
- {
- return store;
- }
- public void setStore(HBaseStore store)
+ @Override
+ protected Scan operationScan()
{
- this.store = store;
+ Scan scan;
+ if (getLastReadRow() == null && getStartRow() == null) {
+ // If no start row specified and no row read yet
+ scan = new Scan();
+ } else if (getEndRow() == null) {
+ // If only start row specified
+ scan = new Scan(Bytes.toBytes(getLastReadRow() == null ? getStartRow() : getLastReadRow()));
+ } else {
+ // If end row also specified
+ scan = new Scan(Bytes.toBytes(getLastReadRow() == null ? getStartRow() : getLastReadRow()),
+ Bytes.toBytes(getEndRow()));
+ }
+ for (HBaseFieldInfo field : tableInfo.getFieldsInfo()) {
+ scan.addColumn(Bytes.toBytes(field.getFamilyName()), Bytes.toBytes(field.getColumnName()));
+ }
+ scan.setAttribute(Scan.HINT_LOOKAHEAD, Bytes.toBytes(getHintScanLookahead()));
+ return scan;
--- End diff --
Can you reuse the scan variable? That will save connection time for every request.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by bhupeshchawda <gi...@git.apache.org>.
Github user bhupeshchawda commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61710530
--- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBasePOJOInputOperator.java ---
@@ -69,116 +69,103 @@ public Object convertValue( HBaseFieldInfo fieldInfo, Object value)
public void setup(OperatorContext context)
{
try {
- store.connect();
+ super.setup(context);
pojoType = Class.forName(pojoTypeName);
pojoType.newInstance(); //try create new instance to verify the class.
rowSetter = PojoUtils.createSetter(pojoType, tableInfo.getRowOrIdExpression(), String.class);
- fieldValueGenerator = FieldValueGenerator.getFieldValueGenerator(pojoType, tableInfo.getFieldsInfo() );
+ fieldValueGenerator = HBaseFieldValueGenerator.getHBaseFieldValueGenerator(pojoType, tableInfo.getFieldsInfo() );
valueConverter = new BytesValueConverter();
} catch (Exception ex) {
throw new RuntimeException(ex);
}
}
@Override
- public void beginWindow(long windowId)
- {
- }
-
- @Override
- public void teardown()
+ protected Object getTuple(Result result)
{
try {
- store.disconnect();
- } catch (IOException ex) {
- throw new RuntimeException(ex);
- }
- }
-
- @Override
- public void emitTuples()
- {
- try {
- Scan scan = nextScan();
- if (scan == null)
- return;
-
- ResultScanner resultScanner = store.getTable().getScanner(scan);
-
- while (true) {
- Result result = resultScanner.next();
- if (result == null)
- break;
-
- String readRow = Bytes.toString(result.getRow());
- if( readRow.equals( lastReadRow ))
- continue;
-
- Object instance = pojoType.newInstance();
- rowSetter.set(instance, readRow);
-
- List<Cell> cells = result.listCells();
+ String readRow = Bytes.toString(result.getRow());
+ if( readRow.equals( getLastReadRow() )) {
+ return null;
+ }
- for (Cell cell : cells) {
- String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
- byte[] value = CellUtil.cloneValue(cell);
- fieldValueGenerator.setColumnValue( instance, columnName, value, valueConverter );
- }
+ Object instance = pojoType.newInstance();
+ rowSetter.set(instance, readRow);
- outputPort.emit(instance);
- lastReadRow = readRow;
+ List<Cell> cells = result.listCells();
+ for (Cell cell : cells) {
+ String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
+ String columnFamily = Bytes.toString(CellUtil.cloneFamily(cell));
+ byte[] value = CellUtil.cloneValue(cell);
+ ((HBaseFieldValueGenerator)fieldValueGenerator).setColumnValue(instance, columnName, columnFamily, value,
+ valueConverter);
}
+ setLastReadRow(readRow);
+ return instance;
} catch (Exception e) {
- throw new RuntimeException(e.getMessage());
+ throw new RuntimeException(e);
}
-
- }
-
- protected Scan nextScan()
- {
- if(lastReadRow==null && startRow==null )
- return new Scan();
- else
- return new Scan( Bytes.toBytes( lastReadRow == null ? startRow : lastReadRow ) );
}
- public HBaseStore getStore()
- {
- return store;
- }
- public void setStore(HBaseStore store)
+ @Override
+ protected Scan operationScan()
{
- this.store = store;
+ Scan scan;
+ if (getLastReadRow() == null && getStartRow() == null) {
+ // If no start row specified and no row read yet
+ scan = new Scan();
+ } else if (getEndRow() == null) {
+ // If only start row specified
+ scan = new Scan(Bytes.toBytes(getLastReadRow() == null ? getStartRow() : getLastReadRow()));
+ } else {
+ // If end row also specified
+ scan = new Scan(Bytes.toBytes(getLastReadRow() == null ? getStartRow() : getLastReadRow()),
+ Bytes.toBytes(getEndRow()));
+ }
+ for (HBaseFieldInfo field : tableInfo.getFieldsInfo()) {
+ scan.addColumn(Bytes.toBytes(field.getFamilyName()), Bytes.toBytes(field.getColumnName()));
+ }
+ scan.setAttribute(Scan.HINT_LOOKAHEAD, Bytes.toBytes(getHintScanLookahead()));
+ return scan;
--- End diff --
done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61074763
--- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBaseScanOperator.java ---
@@ -40,27 +51,106 @@
* @tags hbase, scan, input operator
* @since 0.3.2
*/
-public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T>
+public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T> implements Operator.ActivationListener<Context>
{
+ public static final int DEF_HINT_SCAN_LOOKAHEAD = 2;
+ public static final int DEF_QUEUE_SIZE = 1000;
+ public static final int DEF_SLEEP_MILLIS = 10;
+
+ private String startRow;
+ private String endRow;
+ private String lastReadRow;
+ private int hintScanLookahead = DEF_HINT_SCAN_LOOKAHEAD;
+ private int queueSize = DEF_QUEUE_SIZE;
+ private int sleepMillis = DEF_SLEEP_MILLIS;
+ private Queue<Result> resultQueue;
+
+ @AutoMetric
+ protected long tuplesRead;
+
+ // Transients
+ protected transient Scan scan;
+ protected transient ResultScanner scanner;
+ protected transient Thread readThread;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ resultQueue = Queues.newLinkedBlockingQueue(queueSize);
+ }
+
+ @Override
+ public void activate(Context context)
+ {
+ startReadThread();
+ }
+
+ protected void startReadThread()
+ {
+ try {
+ scan = operationScan();
+ scanner = getStore().getTable().getScanner(scan);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ readThread = new Thread(new Runnable() {
+ @Override
+ public void run()
+ {
+ try {
+ Result result;
+ while ((result = scanner.next()) != null) {
+ while (!resultQueue.offer(result)) {
+ Thread.sleep(sleepMillis);
+ }
+ }
+ } catch (Exception e) {
+ logger.debug("Exception in fetching results {}", e.getMessage());
+ throw new RuntimeException(e);
+ } finally {
+ scanner.close();
+ }
+ }
+ });
+ readThread.start();
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ tuplesRead = 0;
+ }
@Override
public void emitTuples()
{
+ if (!readThread.isAlive() && resultQueue.isEmpty()) {
+ startReadThread();
+ }
try {
- HTable table = getTable();
- Scan scan = operationScan();
- ResultScanner scanner = table.getScanner(scan);
- for (Result result : scanner) {
- //KeyValue[] kvs = result.raw();
- //T t = getTuple(kvs);
- T t = getTuple(result);
- outputPort.emit(t);
+ Result result = resultQueue.poll();
+ if (result == null) {
+ Thread.sleep(sleepMillis);
--- End diff --
Need not sleep here as the engine will handle it if there is no tuple emitted during the call of emitTuples.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by bhupeshchawda <gi...@git.apache.org>.
Github user bhupeshchawda commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61719506
--- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBaseScanOperator.java ---
@@ -40,27 +51,106 @@
* @tags hbase, scan, input operator
* @since 0.3.2
*/
-public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T>
+public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T> implements Operator.ActivationListener<Context>
{
+ public static final int DEF_HINT_SCAN_LOOKAHEAD = 2;
+ public static final int DEF_QUEUE_SIZE = 1000;
+ public static final int DEF_SLEEP_MILLIS = 10;
+
+ private String startRow;
+ private String endRow;
+ private String lastReadRow;
+ private int hintScanLookahead = DEF_HINT_SCAN_LOOKAHEAD;
+ private int queueSize = DEF_QUEUE_SIZE;
+ private int sleepMillis = DEF_SLEEP_MILLIS;
+ private Queue<Result> resultQueue;
+
+ @AutoMetric
+ protected long tuplesRead;
+
+ // Transients
+ protected transient Scan scan;
+ protected transient ResultScanner scanner;
+ protected transient Thread readThread;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ resultQueue = Queues.newLinkedBlockingQueue(queueSize);
+ }
+
+ @Override
+ public void activate(Context context)
+ {
+ startReadThread();
+ }
+
+ protected void startReadThread()
+ {
+ try {
+ scan = operationScan();
+ scanner = getStore().getTable().getScanner(scan);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ readThread = new Thread(new Runnable() {
+ @Override
+ public void run()
+ {
+ try {
+ Result result;
+ while ((result = scanner.next()) != null) {
+ while (!resultQueue.offer(result)) {
+ Thread.sleep(sleepMillis);
+ }
+ }
+ } catch (Exception e) {
+ logger.debug("Exception in fetching results {}", e.getMessage());
+ throw new RuntimeException(e);
+ } finally {
+ scanner.close();
+ }
+ }
+ });
+ readThread.start();
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ tuplesRead = 0;
+ }
@Override
public void emitTuples()
{
+ if (!readThread.isAlive() && resultQueue.isEmpty()) {
+ startReadThread();
+ }
try {
- HTable table = getTable();
- Scan scan = operationScan();
- ResultScanner scanner = table.getScanner(scan);
- for (Result result : scanner) {
- //KeyValue[] kvs = result.raw();
- //T t = getTuple(kvs);
- T t = getTuple(result);
- outputPort.emit(t);
+ Result result = resultQueue.poll();
+ if (result == null) {
+ Thread.sleep(sleepMillis);
--- End diff --
done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by bhupeshchawda <gi...@git.apache.org>.
Github user bhupeshchawda commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61719426
--- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBaseScanOperator.java ---
@@ -40,27 +51,106 @@
* @tags hbase, scan, input operator
* @since 0.3.2
*/
-public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T>
+public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T> implements Operator.ActivationListener<Context>
{
+ public static final int DEF_HINT_SCAN_LOOKAHEAD = 2;
+ public static final int DEF_QUEUE_SIZE = 1000;
+ public static final int DEF_SLEEP_MILLIS = 10;
+
+ private String startRow;
+ private String endRow;
+ private String lastReadRow;
+ private int hintScanLookahead = DEF_HINT_SCAN_LOOKAHEAD;
+ private int queueSize = DEF_QUEUE_SIZE;
+ private int sleepMillis = DEF_SLEEP_MILLIS;
+ private Queue<Result> resultQueue;
+
+ @AutoMetric
+ protected long tuplesRead;
+
+ // Transients
+ protected transient Scan scan;
+ protected transient ResultScanner scanner;
+ protected transient Thread readThread;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ resultQueue = Queues.newLinkedBlockingQueue(queueSize);
+ }
+
+ @Override
+ public void activate(Context context)
+ {
+ startReadThread();
+ }
+
+ protected void startReadThread()
+ {
+ try {
+ scan = operationScan();
+ scanner = getStore().getTable().getScanner(scan);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ readThread = new Thread(new Runnable() {
+ @Override
+ public void run()
+ {
+ try {
+ Result result;
+ while ((result = scanner.next()) != null) {
+ while (!resultQueue.offer(result)) {
+ Thread.sleep(sleepMillis);
+ }
+ }
+ } catch (Exception e) {
+ logger.debug("Exception in fetching results {}", e.getMessage());
+ throw new RuntimeException(e);
+ } finally {
+ scanner.close();
+ }
+ }
+ });
+ readThread.start();
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ tuplesRead = 0;
+ }
@Override
public void emitTuples()
{
+ if (!readThread.isAlive() && resultQueue.isEmpty()) {
+ startReadThread();
--- End diff --
done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-malhar pull request: APEXMALHAR-1957: Added threadi...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/212#discussion_r61074900
--- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBaseScanOperator.java ---
@@ -40,27 +51,106 @@
* @tags hbase, scan, input operator
* @since 0.3.2
*/
-public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T>
+public abstract class HBaseScanOperator<T> extends HBaseInputOperator<T> implements Operator.ActivationListener<Context>
{
+ public static final int DEF_HINT_SCAN_LOOKAHEAD = 2;
+ public static final int DEF_QUEUE_SIZE = 1000;
+ public static final int DEF_SLEEP_MILLIS = 10;
+
+ private String startRow;
+ private String endRow;
+ private String lastReadRow;
+ private int hintScanLookahead = DEF_HINT_SCAN_LOOKAHEAD;
+ private int queueSize = DEF_QUEUE_SIZE;
+ private int sleepMillis = DEF_SLEEP_MILLIS;
+ private Queue<Result> resultQueue;
+
+ @AutoMetric
+ protected long tuplesRead;
+
+ // Transients
+ protected transient Scan scan;
+ protected transient ResultScanner scanner;
+ protected transient Thread readThread;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ resultQueue = Queues.newLinkedBlockingQueue(queueSize);
+ }
+
+ @Override
+ public void activate(Context context)
+ {
+ startReadThread();
+ }
+
+ protected void startReadThread()
+ {
+ try {
+ scan = operationScan();
+ scanner = getStore().getTable().getScanner(scan);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ readThread = new Thread(new Runnable() {
+ @Override
+ public void run()
+ {
+ try {
+ Result result;
+ while ((result = scanner.next()) != null) {
+ while (!resultQueue.offer(result)) {
+ Thread.sleep(sleepMillis);
+ }
+ }
+ } catch (Exception e) {
+ logger.debug("Exception in fetching results {}", e.getMessage());
+ throw new RuntimeException(e);
+ } finally {
+ scanner.close();
+ }
+ }
+ });
+ readThread.start();
+ }
+
+ @Override
+ public void beginWindow(long windowId)
+ {
+ super.beginWindow(windowId);
+ tuplesRead = 0;
+ }
@Override
public void emitTuples()
{
+ if (!readThread.isAlive() && resultQueue.isEmpty()) {
+ startReadThread();
+ }
try {
- HTable table = getTable();
- Scan scan = operationScan();
- ResultScanner scanner = table.getScanner(scan);
- for (Result result : scanner) {
- //KeyValue[] kvs = result.raw();
- //T t = getTuple(kvs);
- T t = getTuple(result);
- outputPort.emit(t);
+ Result result = resultQueue.poll();
+ if (result == null) {
+ Thread.sleep(sleepMillis);
+ return;
+ }
+ T tuple = getTuple(result);
+ if (tuple != null) {
+ outputPort.emit(tuple);
+ tuplesRead++;
}
} catch (Exception e) {
e.printStackTrace();
--- End diff --
Throw RTE.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---