public Scan getHistoryRawTableScan(String cluster, String minJobId,
      String maxJobId, boolean reprocess, boolean includeRaw) {
    Scan scan = new Scan();

    LOG.info("Creating scan for cluster: " + cluster);

    // Add the columns to be pulled back by this scan.
    scan.addFamily(Constants.INFO_FAM_BYTES);
    if (includeRaw) {
      scan.addFamily(Constants.RAW_FAM_BYTES);
    }

    // Pull data only for our cluster
    byte[] clusterPrefix = Bytes.toBytes(cluster + Constants.SEP);
    byte[] startRow;
    if (minJobId == null) {
      startRow = clusterPrefix;
    } else {
      startRow = idConv.toBytes(new QualifiedJobId(cluster, minJobId));
    }
    scan.setStartRow(startRow);

    LOG.info("Starting raw table scan at " + Bytes.toStringBinary(startRow)
        + " " + idConv.fromBytes(startRow));

    FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL);

    // Scan only those raw rows for the specified cluster.
    PrefixFilter prefixFilter = new PrefixFilter(clusterPrefix);
    filters.addFilter(prefixFilter);

    byte[] stopRow;
    // Set to stop the scan once the last row is encountered.
    if (maxJobId != null) {
      // The inclusive stop filter actually is the accurate representation of
      // what needs to be in the result.
      byte[] lastRow = idConv.toBytes(new QualifiedJobId(cluster, maxJobId));
      InclusiveStopFilter inclusiveStopFilter =
          new InclusiveStopFilter(lastRow);
      filters.addFilter(inclusiveStopFilter);
      LOG.info("Stopping raw table scan (stop filter) at "
          + Bytes.toStringBinary(lastRow) + " " + idConv.fromBytes(lastRow));

      // Add one to the jobSequence of the maximum JobId.
      JobId maximumJobId = new JobId(maxJobId);
      JobId oneBiggerThanMaxJobId = new JobId(maximumJobId.getJobEpoch(),
          maximumJobId.getJobSequence() + 1);
      stopRow =
          idConv.toBytes(new QualifiedJobId(cluster, oneBiggerThanMaxJobId));

    } else {
      char oneBiggerSep = (char) (Constants.SEP_CHAR + 1);
      stopRow = Bytes.toBytes(cluster + oneBiggerSep);
    }
    // In addition to InclusiveStopRowFilter, set an estimated end-row that is
    // guaranteed to be bigger than the last row we want (but may over-shoot a
    // little). This helps the TableInput format limit the number of regions
    // (-servers) that need to be targeted for this scan.
    scan.setStopRow(stopRow);
    LOG.info("Stopping raw table scan (stop row) at "
        + Bytes.toStringBinary(stopRow));

    scan.setFilter(filters);

    if (reprocess) {
      SingleColumnValueExcludeFilter columnValueFilter =
          new SingleColumnValueExcludeFilter(Constants.INFO_FAM_BYTES,
              Constants.RAW_COL_REPROCESS_BYTES, CompareFilter.CompareOp.EQUAL,
              Bytes.toBytes(true));
      columnValueFilter.setFilterIfMissing(true);
      filters.addFilter(columnValueFilter);
    } else {
      // Process each row only once. If it is already processed, then do not do
      // it again.
      SingleColumnValueExcludeFilter columnValueFilter =
          new SingleColumnValueExcludeFilter(Constants.INFO_FAM_BYTES,
              Constants.JOB_PROCESSED_SUCCESS_COL_BYTES,
              CompareFilter.CompareOp.NOT_EQUAL, Bytes.toBytes(true));
      filters.addFilter(columnValueFilter);
    }

    /*
     * Create a filter that passes only if both the jobconf and job history
     * blobs are present. Here we use the last mod time columns as a proxy for
     * their existence, since the byte comparison is much cheaper (requires an
     * array copy). We do not want to process rows where we don't have both.
     */
    byte[] empty = Bytes.toBytes(0L);
    FilterList bothColumnFilters =
        new FilterList(FilterList.Operator.MUST_PASS_ALL);
    SingleColumnValueFilter jobConfFilter = new SingleColumnValueFilter(
        Constants.INFO_FAM_BYTES, Constants.JOBCONF_LAST_MODIFIED_COL_BYTES,
        CompareFilter.CompareOp.GREATER, empty);
    jobConfFilter.setFilterIfMissing(true);
    bothColumnFilters.addFilter(jobConfFilter);
    SingleColumnValueFilter jobHistoryFilter = new SingleColumnValueFilter(
        Constants.INFO_FAM_BYTES, Constants.JOBHISTORY_LAST_MODIFIED_COL_BYTES,
        CompareFilter.CompareOp.GREATER, empty);
    jobHistoryFilter.setFilterIfMissing(true);
    bothColumnFilters.addFilter(jobHistoryFilter);

    filters.addFilter(bothColumnFilters);

    scan.setFilter(filters);

    // Let's be nice; we are reading potentially large amounts of data that
    // could take a bit to process.
    scan.setCacheBlocks(false);
    scan.setCaching(1);

    scan.setMaxVersions(1);

    return scan;
  }
public Scan getHistoryRawTableScan ( String cluster , String minJobId , String maxJobId , boolean reprocess , boolean includeRaw ) { Scan scan = new Scan ( ) ; LOG . info ( "Creating scan for cluster: " + cluster ) ; scan . addFamily ( Constants . INFO_FAM_BYTES ) ; if ( includeRaw ) { scan . addFamily ( Constants . RAW_FAM_BYTES ) ; } byte [ ] clusterPrefix = Bytes . toBytes ( cluster + Constants . SEP ) ; byte [ ] startRow ; if ( minJobId == null ) { startRow = clusterPrefix ; } else { startRow = idConv . toBytes ( new QualifiedJobId ( cluster , minJobId ) ) ; } scan . setStartRow ( startRow ) ; LOG . info ( "Starting raw table scan at " + Bytes . toStringBinary ( startRow ) + " " + idConv . fromBytes ( startRow ) ) ; FilterList filters = new FilterList ( FilterList . Operator . MUST_PASS_ALL ) ; PrefixFilter prefixFilter = new PrefixFilter ( clusterPrefix ) ; filters . addFilter ( prefixFilter ) ; byte [ ] stopRow ; if ( maxJobId != null ) { byte [ ] lastRow = idConv . toBytes ( new QualifiedJobId ( cluster , maxJobId ) ) ; InclusiveStopFilter inclusiveStopFilter = new InclusiveStopFilter ( lastRow ) ; filters . addFilter ( inclusiveStopFilter ) ; LOG . info ( "Stopping raw table scan (stop filter) at " + Bytes . toStringBinary ( lastRow ) + " " + idConv . fromBytes ( lastRow ) ) ; JobId maximumJobId = new JobId ( maxJobId ) ; JobId oneBiggerThanMaxJobId = new JobId ( maximumJobId . getJobEpoch ( ) , maximumJobId . getJobSequence ( ) + 1 ) ; stopRow = idConv . toBytes ( new QualifiedJobId ( cluster , oneBiggerThanMaxJobId ) ) ; } else { char oneBiggerSep = ( char ) ( Constants . SEP_CHAR + 1 ) ; stopRow = Bytes . toBytes ( cluster + oneBiggerSep ) ; } scan . setStopRow ( stopRow ) ; LOG . info ( "Stopping raw table scan (stop row) at " + Bytes . toStringBinary ( stopRow ) ) ; scan . setFilter ( filters ) ; if ( reprocess ) { SingleColumnValueExcludeFilter columnValueFilter = new SingleColumnValueExcludeFilter ( Constants . INFO_FAM_BYTES , Constants . RAW_COL_REPROCESS_BYTES , CompareFilter . CompareOp . EQUAL , Bytes . toBytes ( true ) ) ; columnValueFilter . setFilterIfMissing ( true ) ; filters . addFilter ( columnValueFilter ) ; } else { SingleColumnValueExcludeFilter columnValueFilter = new SingleColumnValueExcludeFilter ( Constants . INFO_FAM_BYTES , Constants . JOB_PROCESSED_SUCCESS_COL_BYTES , CompareFilter . CompareOp . NOT_EQUAL , Bytes . toBytes ( true ) ) ; filters . addFilter ( columnValueFilter ) ; } byte [ ] empty = Bytes . toBytes ( 0L ) ; FilterList bothColumnFilters = new FilterList ( FilterList . Operator . MUST_PASS_ALL ) ; SingleColumnValueFilter jobConfFilter = new SingleColumnValueFilter ( Constants . INFO_FAM_BYTES , Constants . JOBCONF_LAST_MODIFIED_COL_BYTES , CompareFilter . CompareOp . GREATER , empty ) ; jobConfFilter . setFilterIfMissing ( true ) ; bothColumnFilters . addFilter ( jobConfFilter ) ; SingleColumnValueFilter jobHistoryFilter = new SingleColumnValueFilter ( Constants . INFO_FAM_BYTES , Constants . JOBHISTORY_LAST_MODIFIED_COL_BYTES , CompareFilter . CompareOp . GREATER , empty ) ; jobHistoryFilter . setFilterIfMissing ( true ) ; bothColumnFilters . addFilter ( jobHistoryFilter ) ; filters . addFilter ( bothColumnFilters ) ; scan . setFilter ( filters ) ; scan . setCacheBlocks ( false ) ; scan . setCaching ( 1 ) ; scan . setMaxVersions ( 1 ) ; return scan ; }
Get a {@link Scan} to go through all the records loaded in the
{@link Constants#HISTORY_RAW_TABLE} that match the given parameters.

@param cluster on which the Hadoop jobs ran.
@param minJobId used to start the scan. If null then there is no min limit
on JobId.
@param maxJobId used to end the scan (inclusive). If null then there is no
max limit on jobId.
@param reprocess return only those raw records that were marked to be
reprocessed using
{@link #markJobForReprocesssing(QualifiedJobId)}. Otherwise
successfully processed jobs are skipped.
@param reprocess When true then reprocess argument is ignored and is
assumed to be true.
@param includeRaw whether to include the raw column family in the scan
results.
@return a scan of jobIds between the specified min and max. Retrieves only
one version of each column.