@Test public void testGetFlowTimeSeriesStats() throws Exception { final short numJobsAppOne = 3; final short numJobsAppTwo = 4; final long baseStats = 10L; flowDataGen.loadFlow("c1@local", "buser", "AppOne", 1234, "a", numJobsAppOne, baseStats, idService, historyTable); flowDataGen.loadFlow("c1@local", "buser", "AppTwo", 2345, "b", numJobsAppTwo, baseStats, idService, historyTable); JobHistoryService service = null; service = new JobHistoryService(UTIL.getConfiguration(), hbaseConnection); List<Flow> flowSeries = service.getFlowTimeSeriesStats("c1@local", "buser", "AppOne", "", 0L, 0L, 1000, null); checkSomeFlowStats("a", HadoopVersion.ONE, numJobsAppOne, baseStats, flowSeries); flowSeries = service.getFlowTimeSeriesStats("c1@local", "buser", "AppTwo", "", 0L, 0L, 1000, null); checkSomeFlowStats("b", HadoopVersion.ONE, numJobsAppTwo, baseStats, flowSeries); }
public List<Flow> getFlowTimeSeriesStats(String cluster, String user, String appId, String version, long startTime, long endTime, int limit, byte[] startRow) throws IOException { byte[] rowPrefix = Bytes.toBytes((cluster + Constants.SEP + user + Constants.SEP + appId + Constants.SEP)); byte[] scanStartRow; if (startRow != null) { scanStartRow = startRow; } else { if (endTime != 0) { long endRunId = FlowKey.encodeRunId(endTime); scanStartRow = Bytes.add(rowPrefix, Bytes.toBytes(endRunId), Constants.SEP_BYTES); } else { scanStartRow = rowPrefix; } } Scan scan = new Scan(); scan.setStartRow(scanStartRow); FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL); if (startTime != 0) { long startRunId = FlowKey.encodeRunId(startTime); byte[] scanEndRow = Bytes.add(rowPrefix, Bytes.toBytes(startRunId), Constants.ZERO_SINGLE_BYTE); scan.setStopRow(scanEndRow); } else { filters.addFilter(new WhileMatchFilter(new PrefixFilter(rowPrefix))); } if (version != null && version.length() > 0) { filters.addFilter(new SingleColumnValueFilter(Constants.INFO_FAM_BYTES, Constants.VERSION_COLUMN_BYTES, CompareFilter.CompareOp.EQUAL, Bytes.toBytes(version))); } filters.addFilter(new QualifierFilter(CompareFilter.CompareOp.NOT_EQUAL, new RegexStringComparator( "^c\\!((?!" + Constants.HRAVEN_QUEUE + ").)*$"))); scan.setFilter(filters); LOG.info("scan : \n " + scan.toJSON() + " \n"); return createFromResults(scan, false, limit); }
JobHistoryService { public List<Flow> getFlowTimeSeriesStats(String cluster, String user, String appId, String version, long startTime, long endTime, int limit, byte[] startRow) throws IOException { byte[] rowPrefix = Bytes.toBytes((cluster + Constants.SEP + user + Constants.SEP + appId + Constants.SEP)); byte[] scanStartRow; if (startRow != null) { scanStartRow = startRow; } else { if (endTime != 0) { long endRunId = FlowKey.encodeRunId(endTime); scanStartRow = Bytes.add(rowPrefix, Bytes.toBytes(endRunId), Constants.SEP_BYTES); } else { scanStartRow = rowPrefix; } } Scan scan = new Scan(); scan.setStartRow(scanStartRow); FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL); if (startTime != 0) { long startRunId = FlowKey.encodeRunId(startTime); byte[] scanEndRow = Bytes.add(rowPrefix, Bytes.toBytes(startRunId), Constants.ZERO_SINGLE_BYTE); scan.setStopRow(scanEndRow); } else { filters.addFilter(new WhileMatchFilter(new PrefixFilter(rowPrefix))); } if (version != null && version.length() > 0) { filters.addFilter(new SingleColumnValueFilter(Constants.INFO_FAM_BYTES, Constants.VERSION_COLUMN_BYTES, CompareFilter.CompareOp.EQUAL, Bytes.toBytes(version))); } filters.addFilter(new QualifierFilter(CompareFilter.CompareOp.NOT_EQUAL, new RegexStringComparator( "^c\\!((?!" + Constants.HRAVEN_QUEUE + ").)*$"))); scan.setFilter(filters); LOG.info("scan : \n " + scan.toJSON() + " \n"); return createFromResults(scan, false, limit); } }
JobHistoryService { public List<Flow> getFlowTimeSeriesStats(String cluster, String user, String appId, String version, long startTime, long endTime, int limit, byte[] startRow) throws IOException { byte[] rowPrefix = Bytes.toBytes((cluster + Constants.SEP + user + Constants.SEP + appId + Constants.SEP)); byte[] scanStartRow; if (startRow != null) { scanStartRow = startRow; } else { if (endTime != 0) { long endRunId = FlowKey.encodeRunId(endTime); scanStartRow = Bytes.add(rowPrefix, Bytes.toBytes(endRunId), Constants.SEP_BYTES); } else { scanStartRow = rowPrefix; } } Scan scan = new Scan(); scan.setStartRow(scanStartRow); FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL); if (startTime != 0) { long startRunId = FlowKey.encodeRunId(startTime); byte[] scanEndRow = Bytes.add(rowPrefix, Bytes.toBytes(startRunId), Constants.ZERO_SINGLE_BYTE); scan.setStopRow(scanEndRow); } else { filters.addFilter(new WhileMatchFilter(new PrefixFilter(rowPrefix))); } if (version != null && version.length() > 0) { filters.addFilter(new SingleColumnValueFilter(Constants.INFO_FAM_BYTES, Constants.VERSION_COLUMN_BYTES, CompareFilter.CompareOp.EQUAL, Bytes.toBytes(version))); } filters.addFilter(new QualifierFilter(CompareFilter.CompareOp.NOT_EQUAL, new RegexStringComparator( "^c\\!((?!" + Constants.HRAVEN_QUEUE + ").)*$"))); scan.setFilter(filters); LOG.info("scan : \n " + scan.toJSON() + " \n"); return createFromResults(scan, false, limit); } JobHistoryService(Configuration hbaseConf, Connection hbaseConnection); }
JobHistoryService { public List<Flow> getFlowTimeSeriesStats(String cluster, String user, String appId, String version, long startTime, long endTime, int limit, byte[] startRow) throws IOException { byte[] rowPrefix = Bytes.toBytes((cluster + Constants.SEP + user + Constants.SEP + appId + Constants.SEP)); byte[] scanStartRow; if (startRow != null) { scanStartRow = startRow; } else { if (endTime != 0) { long endRunId = FlowKey.encodeRunId(endTime); scanStartRow = Bytes.add(rowPrefix, Bytes.toBytes(endRunId), Constants.SEP_BYTES); } else { scanStartRow = rowPrefix; } } Scan scan = new Scan(); scan.setStartRow(scanStartRow); FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL); if (startTime != 0) { long startRunId = FlowKey.encodeRunId(startTime); byte[] scanEndRow = Bytes.add(rowPrefix, Bytes.toBytes(startRunId), Constants.ZERO_SINGLE_BYTE); scan.setStopRow(scanEndRow); } else { filters.addFilter(new WhileMatchFilter(new PrefixFilter(rowPrefix))); } if (version != null && version.length() > 0) { filters.addFilter(new SingleColumnValueFilter(Constants.INFO_FAM_BYTES, Constants.VERSION_COLUMN_BYTES, CompareFilter.CompareOp.EQUAL, Bytes.toBytes(version))); } filters.addFilter(new QualifierFilter(CompareFilter.CompareOp.NOT_EQUAL, new RegexStringComparator( "^c\\!((?!" + Constants.HRAVEN_QUEUE + ").)*$"))); scan.setFilter(filters); LOG.info("scan : \n " + scan.toJSON() + " \n"); return createFromResults(scan, false, limit); } JobHistoryService(Configuration hbaseConf, Connection hbaseConnection); Flow getLatestFlow(String cluster, String user, String appId); Flow getLatestFlow(String cluster, String user, String appId,
      boolean populateTasks); List<Flow> getFlowSeries(String cluster, String user, String appId,
      int limit); Flow getFlow(String cluster, String user, String appId, long runId,
      boolean populateTasks); Flow getFlowByJobID(String cluster, String jobId,
      boolean populateTasks); List<Flow> getFlowSeries(String cluster, String user, String appId,
      String version, boolean populateTasks, int limit); List<Flow> getFlowSeries(String cluster, String user, String appId,
      String version, boolean populateTasks, long startTime, long endTime,
      int limit); List<Flow> getFlowTimeSeriesStats(String cluster, String user,
      String appId, String version, long startTime, long endTime, int limit,
      byte[] startRow); JobDetails getJobByJobID(String cluster, String jobId); JobDetails getJobByJobID(String cluster, String jobId,
      boolean populateTasks); JobDetails getJobByJobID(QualifiedJobId jobId, boolean populateTasks); static Configuration parseConfiguration(
      Map<byte[], byte[]> keyValues); static CounterMap parseCounters(byte[] prefix,
      Map<byte[], byte[]> keyValues); static List<Put> getHbasePuts(JobDesc jobDesc, Configuration jobConf); int removeJob(JobKey key); }
JobHistoryService { public List<Flow> getFlowTimeSeriesStats(String cluster, String user, String appId, String version, long startTime, long endTime, int limit, byte[] startRow) throws IOException { byte[] rowPrefix = Bytes.toBytes((cluster + Constants.SEP + user + Constants.SEP + appId + Constants.SEP)); byte[] scanStartRow; if (startRow != null) { scanStartRow = startRow; } else { if (endTime != 0) { long endRunId = FlowKey.encodeRunId(endTime); scanStartRow = Bytes.add(rowPrefix, Bytes.toBytes(endRunId), Constants.SEP_BYTES); } else { scanStartRow = rowPrefix; } } Scan scan = new Scan(); scan.setStartRow(scanStartRow); FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL); if (startTime != 0) { long startRunId = FlowKey.encodeRunId(startTime); byte[] scanEndRow = Bytes.add(rowPrefix, Bytes.toBytes(startRunId), Constants.ZERO_SINGLE_BYTE); scan.setStopRow(scanEndRow); } else { filters.addFilter(new WhileMatchFilter(new PrefixFilter(rowPrefix))); } if (version != null && version.length() > 0) { filters.addFilter(new SingleColumnValueFilter(Constants.INFO_FAM_BYTES, Constants.VERSION_COLUMN_BYTES, CompareFilter.CompareOp.EQUAL, Bytes.toBytes(version))); } filters.addFilter(new QualifierFilter(CompareFilter.CompareOp.NOT_EQUAL, new RegexStringComparator( "^c\\!((?!" + Constants.HRAVEN_QUEUE + ").)*$"))); scan.setFilter(filters); LOG.info("scan : \n " + scan.toJSON() + " \n"); return createFromResults(scan, false, limit); } JobHistoryService(Configuration hbaseConf, Connection hbaseConnection); Flow getLatestFlow(String cluster, String user, String appId); Flow getLatestFlow(String cluster, String user, String appId,
      boolean populateTasks); List<Flow> getFlowSeries(String cluster, String user, String appId,
      int limit); Flow getFlow(String cluster, String user, String appId, long runId,
      boolean populateTasks); Flow getFlowByJobID(String cluster, String jobId,
      boolean populateTasks); List<Flow> getFlowSeries(String cluster, String user, String appId,
      String version, boolean populateTasks, int limit); List<Flow> getFlowSeries(String cluster, String user, String appId,
      String version, boolean populateTasks, long startTime, long endTime,
      int limit); List<Flow> getFlowTimeSeriesStats(String cluster, String user,
      String appId, String version, long startTime, long endTime, int limit,
      byte[] startRow); JobDetails getJobByJobID(String cluster, String jobId); JobDetails getJobByJobID(String cluster, String jobId,
      boolean populateTasks); JobDetails getJobByJobID(QualifiedJobId jobId, boolean populateTasks); static Configuration parseConfiguration(
      Map<byte[], byte[]> keyValues); static CounterMap parseCounters(byte[] prefix,
      Map<byte[], byte[]> keyValues); static List<Put> getHbasePuts(JobDesc jobDesc, Configuration jobConf); int removeJob(JobKey key);  }