content: package com.alibaba.datax.plugin.reader.hbase11xreader;
  
  import com.alibaba.datax.common.element.*;
  import com.alibaba.datax.common.exception.DataXException;
  import org.apache.commons.lang.ArrayUtils;
  import org.apache.commons.lang3.time.DateUtils;
  import org.apache.hadoop.hbase.client.Result;
  import org.apache.hadoop.hbase.client.ResultScanner;
  import org.apache.hadoop.hbase.client.Scan;
  import org.apache.hadoop.hbase.client.Table;
  import org.apache.hadoop.hbase.util.Bytes;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  
  public abstract class HbaseAbstractTask {
      private final static Logger LOG = LoggerFactory.getLogger(HbaseAbstractTask.class);
  
      private byte[] startKey = null;
      private byte[] endKey = null;
  
      protected Table htable;
      protected String encoding;
      protected int scanCacheSize;
      protected int  scanBatchSize;
  
      protected Result lastResult = null;
      protected Scan scan;
      protected ResultScanner resultScanner;
  
  
      public HbaseAbstractTask(com.alibaba.datax.common.util.Configuration configuration) {
  
          this.htable = Hbase11xHelper.getTable(configuration);
  
          this.encoding = configuration.getString(Key.ENCODING,Constant.DEFAULT_ENCODING);
          this.startKey = Hbase11xHelper.convertInnerStartRowkey(configuration);
          this.endKey =  Hbase11xHelper.convertInnerEndRowkey(configuration);
          this.scanCacheSize = configuration.getInt(Key.SCAN_CACHE_SIZE,Constant.DEFAULT_SCAN_CACHE_SIZE);
          this.scanBatchSize = configuration.getInt(Key.SCAN_BATCH_SIZE,Constant.DEFAULT_SCAN_BATCH_SIZE);
      }
  
      public abstract boolean fetchLine(Record record) throws Exception;
  
      //不同模式设置不同,如多版本模式需要设置版本
      public abstract void initScan(Scan scan);
  
  
      public void prepare() throws Exception {
          this.scan = new Scan();
          this.scan.setSmall(false);
          this.scan.setStartRow(startKey);
          this.scan.setStopRow(endKey);
          LOG.info("The task set startRowkey=[{}], endRowkey=[{}].", Bytes.toStringBinary(this.startKey), Bytes.toStringBinary(this.endKey));
          //scan的Caching Batch全部留在hconfig中每次从服务器端读取的行数，设置默认值未256
          this.scan.setCaching(this.scanCacheSize);
          //设置获取记录的列个数，hbase默认无限制，也就是返回所有的列,这里默认是100
          this.scan.setBatch(this.scanBatchSize);
          //为是否缓存块，hbase默认缓存,同步全部数据时非热点数据，因此不需要缓存
          this.scan.setCacheBlocks(false);
          initScan(this.scan);
  
          this.resultScanner = this.htable.getScanner(this.scan);
      }
  
      public void close()  {
          Hbase11xHelper.closeResultScanner(this.resultScanner);
          Hbase11xHelper.closeTable(this.htable);
      }
  
      protected Result getNextHbaseRow() throws IOException {
          Result result;
          try {
              result = resultScanner.next();
          } catch (IOException e) {
              if (lastResult != null) {
                  this.scan.setStartRow(lastResult.getRow());
              }
              resultScanner = this.htable.getScanner(scan);
              result = resultScanner.next();
              if (lastResult != null && Bytes.equals(lastResult.getRow(), result.getRow())) {
                  result = resultScanner.next();
              }
          }
          lastResult = result;
          // may be null
          return result;
      }
  
      public Column convertBytesToAssignType(ColumnType columnType, byte[] byteArray,String dateformat) throws Exception {
          Column column;
          switch (columnType) {
              case BOOLEAN:
                  column = new BoolColumn(ArrayUtils.isEmpty(byteArray) ? null : Bytes.toBoolean(byteArray));
                  break;
              case SHORT:
                  column = new LongColumn(ArrayUtils.isEmpty(byteArray) ? null : String.valueOf(Bytes.toShort(byteArray)));
                  break;
              case INT:
                  column = new LongColumn(ArrayUtils.isEmpty(byteArray) ? null : Bytes.toInt(byteArray));
                  break;
              case LONG:
                  column = new LongColumn(ArrayUtils.isEmpty(byteArray) ? null : Bytes.toLong(byteArray));
                  break;
              case FLOAT:
                  column = new DoubleColumn(ArrayUtils.isEmpty(byteArray) ? null : Bytes.toFloat(byteArray));
                  break;
              case DOUBLE:
                  column = new DoubleColumn(ArrayUtils.isEmpty(byteArray) ? null : Bytes.toDouble(byteArray));
                  break;
              case STRING:
                  column = new StringColumn(ArrayUtils.isEmpty(byteArray) ? null : new String(byteArray, encoding));
                  break;
              case BINARY_STRING:
                  column = new StringColumn(ArrayUtils.isEmpty(byteArray) ? null : Bytes.toStringBinary(byteArray));
                  break;
              case DATE:
                  String dateValue = Bytes.toStringBinary(byteArray);
                  column = new DateColumn(ArrayUtils.isEmpty(byteArray) ? null : DateUtils.parseDate(dateValue, new String[]{dateformat}));
                  break;
              default:
                  throw DataXException.asDataXException(Hbase11xReaderErrorCode.ILLEGAL_VALUE, "Hbasereader 不支持您配置的列类型:" + columnType);
          }
          return column;
      }
  
      public Column convertValueToAssignType(ColumnType columnType, String constantValue,String dateformat) throws Exception {
          Column column;
          switch (columnType) {
              case BOOLEAN:
                  column = new BoolColumn(constantValue);
                  break;
              case SHORT:
              case INT:
              case LONG:
                  column = new LongColumn(constantValue);
                  break;
              case FLOAT:
              case DOUBLE:
                  column = new DoubleColumn(constantValue);
                  break;
              case STRING:
                  column = new StringColumn(constantValue);
                  break;
              case DATE:
                  column = new DateColumn(DateUtils.parseDate(constantValue, new String[]{dateformat}));
                  break;
              default:
                  throw DataXException.asDataXException(Hbase11xReaderErrorCode.ILLEGAL_VALUE, "Hbasereader 常量列不支持您配置的列类型:" + columnType);
          }
          return column;
      }
  }
  