apex-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (APEXMALHAR-1957) Improve HBasePOJOInputOperator with support for threaded read
Date Thu, 23 Jun 2016 05:17:16 GMT

    [ https://issues.apache.org/jira/browse/APEXMALHAR-1957?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15345758#comment-15345758
] 

ASF GitHub Bot commented on APEXMALHAR-1957:
--------------------------------------------

Github user bhupeshchawda commented on a diff in the pull request:

    https://github.com/apache/apex-malhar/pull/292#discussion_r68176757
  
    --- Diff: contrib/src/main/java/com/datatorrent/contrib/hbase/HBaseScanOperator.java ---
    @@ -40,27 +51,107 @@
      * @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;
    +  private boolean threadFailed = false;
    +
    +  @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());
    +          threadFailed = true;
    +          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() && threadFailed) {
    +      throw new RuntimeException("Exception in scan thread");
    +    }
         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) {
    +        return;
    +      }
    +      T tuple = getTuple(result);
    +      if (tuple != null) {
    +        outputPort.emit(tuple);
    +        tuplesRead++;
           }
         } catch (Exception e) {
    -      e.printStackTrace();
    +      throw new RuntimeException(e);
         }
       }
     
    +  @Override
    +  public void deactivate()
    +  {
    +    readThread.interrupt();
    --- End diff --
    
    I have introduced a control variable for running and pausing the thread. However, the
interrupt in deactivate would not cause any problems, since it is not affecting the state
of the application at all. In normal functioning, we never stop the running thread, just pause
it.


> Improve HBasePOJOInputOperator with support for threaded read
> -------------------------------------------------------------
>
>                 Key: APEXMALHAR-1957
>                 URL: https://issues.apache.org/jira/browse/APEXMALHAR-1957
>             Project: Apache Apex Malhar
>          Issue Type: Task
>            Reporter: Bhupesh Chawda
>            Assignee: Bhupesh Chawda
>
> Add the following support to Hbase POJO Input Operator:
> * Add support for threaded read
> * Allow to specify a set of "column family: column" and fetch data only for these columns
> * Allow to specify an end row key to stop scanning
> * Add metrics



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message