@Test public void TestGetAllDirs() throws IOException { HdfsStatsService hs = new HdfsStatsService(UTIL.getConfiguration(), hbaseConnection); long ts = 1392217200L; String cluster1 = "getAllDirsCluster1"; String pathPrefix = "/all_dirs_dir1/"; long encodedRunId = (Long.MAX_VALUE - ts); loadHdfsData(encodedRunId, cluster1, pathPrefix); List<HdfsStats> h1 = hs.getAllDirs(cluster1, pathPrefix, 10, ts); assertEquals(h1.size(), testDataSize); for (int i = 0; i < h1.size(); i++) { HdfsStats h2 = h1.get(i); assertHdfsStats(h2, cluster1, ts); } List<HdfsStats> h2 = hs.getAllDirs(cluster1, pathPrefix, 2, ts); assertEquals(h2.size(), 2); List<HdfsStats> h3 = hs.getAllDirs(cluster1, "nonexistentpath", 100, ts); assertEquals(h3.size(), 0); }
public List<HdfsStats> getAllDirs(String cluster, String pathPrefix, int limit, long runId) throws IOException { long encodedRunId = getEncodedRunId(runId); String rowPrefixStr = Long.toString(encodedRunId) + HdfsConstants.SEP + cluster; if (StringUtils.isNotEmpty(pathPrefix)) { rowPrefixStr += HdfsConstants.SEP + StringUtil.cleanseToken(pathPrefix); } LOG.info(" Getting all dirs for cluster " + cluster + " with pathPrefix: " + pathPrefix + " for runId " + runId + " encodedRunId: " + encodedRunId + " limit: " + limit + " row prefix : " + rowPrefixStr); byte[] rowPrefix = Bytes.toBytes(rowPrefixStr); Scan scan = createScanWithAllColumns(); scan.setStartRow(rowPrefix); Filter prefixFilter = new WhileMatchFilter(new PrefixFilter(rowPrefix)); scan.setFilter(prefixFilter); scan.setCaching(Math.min(limit, defaultScannerCaching)); scan.setMaxVersions(1); return createFromScanResults(cluster, null, scan, limit, Boolean.FALSE, 0l, 0l); }
HdfsStatsService { public List<HdfsStats> getAllDirs(String cluster, String pathPrefix, int limit, long runId) throws IOException { long encodedRunId = getEncodedRunId(runId); String rowPrefixStr = Long.toString(encodedRunId) + HdfsConstants.SEP + cluster; if (StringUtils.isNotEmpty(pathPrefix)) { rowPrefixStr += HdfsConstants.SEP + StringUtil.cleanseToken(pathPrefix); } LOG.info(" Getting all dirs for cluster " + cluster + " with pathPrefix: " + pathPrefix + " for runId " + runId + " encodedRunId: " + encodedRunId + " limit: " + limit + " row prefix : " + rowPrefixStr); byte[] rowPrefix = Bytes.toBytes(rowPrefixStr); Scan scan = createScanWithAllColumns(); scan.setStartRow(rowPrefix); Filter prefixFilter = new WhileMatchFilter(new PrefixFilter(rowPrefix)); scan.setFilter(prefixFilter); scan.setCaching(Math.min(limit, defaultScannerCaching)); scan.setMaxVersions(1); return createFromScanResults(cluster, null, scan, limit, Boolean.FALSE, 0l, 0l); } }
HdfsStatsService { public List<HdfsStats> getAllDirs(String cluster, String pathPrefix, int limit, long runId) throws IOException { long encodedRunId = getEncodedRunId(runId); String rowPrefixStr = Long.toString(encodedRunId) + HdfsConstants.SEP + cluster; if (StringUtils.isNotEmpty(pathPrefix)) { rowPrefixStr += HdfsConstants.SEP + StringUtil.cleanseToken(pathPrefix); } LOG.info(" Getting all dirs for cluster " + cluster + " with pathPrefix: " + pathPrefix + " for runId " + runId + " encodedRunId: " + encodedRunId + " limit: " + limit + " row prefix : " + rowPrefixStr); byte[] rowPrefix = Bytes.toBytes(rowPrefixStr); Scan scan = createScanWithAllColumns(); scan.setStartRow(rowPrefix); Filter prefixFilter = new WhileMatchFilter(new PrefixFilter(rowPrefix)); scan.setFilter(prefixFilter); scan.setCaching(Math.min(limit, defaultScannerCaching)); scan.setMaxVersions(1); return createFromScanResults(cluster, null, scan, limit, Boolean.FALSE, 0l, 0l); } HdfsStatsService(Configuration hbaseConf, Connection hbaseConnection); }
HdfsStatsService { public List<HdfsStats> getAllDirs(String cluster, String pathPrefix, int limit, long runId) throws IOException { long encodedRunId = getEncodedRunId(runId); String rowPrefixStr = Long.toString(encodedRunId) + HdfsConstants.SEP + cluster; if (StringUtils.isNotEmpty(pathPrefix)) { rowPrefixStr += HdfsConstants.SEP + StringUtil.cleanseToken(pathPrefix); } LOG.info(" Getting all dirs for cluster " + cluster + " with pathPrefix: " + pathPrefix + " for runId " + runId + " encodedRunId: " + encodedRunId + " limit: " + limit + " row prefix : " + rowPrefixStr); byte[] rowPrefix = Bytes.toBytes(rowPrefixStr); Scan scan = createScanWithAllColumns(); scan.setStartRow(rowPrefix); Filter prefixFilter = new WhileMatchFilter(new PrefixFilter(rowPrefix)); scan.setFilter(prefixFilter); scan.setCaching(Math.min(limit, defaultScannerCaching)); scan.setMaxVersions(1); return createFromScanResults(cluster, null, scan, limit, Boolean.FALSE, 0l, 0l); } HdfsStatsService(Configuration hbaseConf, Connection hbaseConnection); static long getEncodedRunId(long now); List<HdfsStats> getAllDirs(String cluster, String pathPrefix,
      int limit, long runId); static long getOlderRunId(int i, long runId); List<HdfsStats> getHdfsTimeSeriesStats(String cluster, String path,
      int limit, long starttime, long endtime); }
HdfsStatsService { public List<HdfsStats> getAllDirs(String cluster, String pathPrefix, int limit, long runId) throws IOException { long encodedRunId = getEncodedRunId(runId); String rowPrefixStr = Long.toString(encodedRunId) + HdfsConstants.SEP + cluster; if (StringUtils.isNotEmpty(pathPrefix)) { rowPrefixStr += HdfsConstants.SEP + StringUtil.cleanseToken(pathPrefix); } LOG.info(" Getting all dirs for cluster " + cluster + " with pathPrefix: " + pathPrefix + " for runId " + runId + " encodedRunId: " + encodedRunId + " limit: " + limit + " row prefix : " + rowPrefixStr); byte[] rowPrefix = Bytes.toBytes(rowPrefixStr); Scan scan = createScanWithAllColumns(); scan.setStartRow(rowPrefix); Filter prefixFilter = new WhileMatchFilter(new PrefixFilter(rowPrefix)); scan.setFilter(prefixFilter); scan.setCaching(Math.min(limit, defaultScannerCaching)); scan.setMaxVersions(1); return createFromScanResults(cluster, null, scan, limit, Boolean.FALSE, 0l, 0l); } HdfsStatsService(Configuration hbaseConf, Connection hbaseConnection); static long getEncodedRunId(long now); List<HdfsStats> getAllDirs(String cluster, String pathPrefix,
      int limit, long runId); static long getOlderRunId(int i, long runId); List<HdfsStats> getHdfsTimeSeriesStats(String cluster, String path,
      int limit, long starttime, long endtime);  }