@Test public void testGetFilesAtPath() throws IOException { String datePattern = "yyyy/MM/dd/HH"; DateTimeFormatter formatter = DateTimeFormat.forPattern(datePattern); LocalDateTime endDate = LocalDateTime.now(DateTimeZone.forID(TimeAwareRecursiveCopyableDataset.DEFAULT_DATE_PATTERN_TIMEZONE)); Set<String> candidateFiles = new HashSet<>(); for (int i = 0; i < MAX_NUM_HOURLY_DIRS; i++) { String startDate = endDate.minusHours(i).toString(formatter); Path subDirPath = new Path(baseDir1, new Path(startDate)); fs.mkdirs(subDirPath); Path filePath = new Path(subDirPath, i + ".avro"); fs.create(filePath); if (i < (NUM_LOOKBACK_HOURS + 1)) { candidateFiles.add(filePath.toString()); } } Properties properties = new Properties(); properties.setProperty(TimeAwareRecursiveCopyableDataset.LOOKBACK_TIME_KEY, NUM_LOOKBACK_HOURS_STR); properties.setProperty(TimeAwareRecursiveCopyableDataset.DATE_PATTERN_KEY, "yyyy/MM/dd/HH"); PathFilter pathFilter = new HiddenFilter(); TimeAwareRecursiveCopyableDataset dataset = new TimeAwareRecursiveCopyableDataset(fs, baseDir1, properties, new Path("/tmp/srchourly")); fileStatusList = dataset.getFilesAtPath(fs, baseDir1, pathFilter); candidateFiles = new HashSet<>(); datePattern = "yyyy/MM/dd/HH"; formatter = DateTimeFormat.forPattern(datePattern); for (int i = 0; i < MAX_NUM_HOURLY_DIRS; i++) { String startDate = endDate.minusHours(i).toString(formatter); Path subDirPath = new Path(baseDir1, new Path(startDate)); Path filePath = new Path(subDirPath, i + ".avro"); if (i < NUM_DAYS_HOURS_DIRS + 1) { candidateFiles.add(filePath.toString()); } } Assert.assertEquals(fileStatusList.size(), NUM_DAYS_HOURS_DIRS + 1); for (FileStatus fileStatus: fileStatusList) { Assert.assertTrue(candidateFiles.contains(PathUtils.getPathWithoutSchemeAndAuthority(fileStatus.getPath()).toString())); } datePattern = "yyyy/MM/dd"; formatter = DateTimeFormat.forPattern(datePattern); endDate = LocalDateTime.now(DateTimeZone.forID(TimeAwareRecursiveCopyableDataset.DEFAULT_DATE_PATTERN_TIMEZONE)); candidateFiles = new HashSet<>(); for (int i = 0; i < MAX_NUM_DAILY_DIRS; i++) { String startDate = endDate.minusDays(i).toString(formatter); Path subDirPath = new Path(baseDir2, new Path(startDate)); fs.mkdirs(subDirPath); Path filePath = new Path(subDirPath, i + ".avro"); fs.create(filePath); if (i < (NUM_LOOKBACK_DAYS + 1)) { candidateFiles.add(filePath.toString()); } } properties = new Properties(); properties.setProperty(TimeAwareRecursiveCopyableDataset.LOOKBACK_TIME_KEY, NUM_LOOKBACK_DAYS_STR); properties.setProperty(TimeAwareRecursiveCopyableDataset.DATE_PATTERN_KEY, "yyyy/MM/dd"); dataset = new TimeAwareRecursiveCopyableDataset(fs, baseDir2, properties, new Path("/tmp/src/*/daily")); fileStatusList = dataset.getFilesAtPath(fs, baseDir2, pathFilter); Assert.assertEquals(fileStatusList.size(), NUM_LOOKBACK_DAYS + 1); for (FileStatus fileStatus: fileStatusList) { Assert.assertTrue(candidateFiles.contains(PathUtils.getPathWithoutSchemeAndAuthority(fileStatus.getPath()).toString())); } datePattern = "yyyy-MM-dd-HH-mm"; formatter = DateTimeFormat.forPattern(datePattern); endDate = LocalDateTime.now(DateTimeZone.forID(TimeAwareRecursiveCopyableDataset.DEFAULT_DATE_PATTERN_TIMEZONE)); Random random = new Random(); candidateFiles = new HashSet<>(); for (int i = 0; i < MAX_NUM_DAILY_DIRS; i++) { String startDate = endDate.minusDays(i).withMinuteOfHour(random.nextInt(60)).toString(formatter); if (i == 0) { startDate = endDate.minusHours(i).withMinuteOfHour(0).toString(formatter); } Path subDirPath = new Path(baseDir3, new Path(startDate)); fs.mkdirs(subDirPath); Path filePath = new Path(subDirPath, i + ".avro"); fs.create(filePath); if (i < (NUM_LOOKBACK_DAYS + 1)) { candidateFiles.add(filePath.toString()); } } properties = new Properties(); properties.setProperty(TimeAwareRecursiveCopyableDataset.LOOKBACK_TIME_KEY, "2d1h"); properties.setProperty(TimeAwareRecursiveCopyableDataset.DATE_PATTERN_KEY, "yyyy-MM-dd-HH-mm"); dataset = new TimeAwareRecursiveCopyableDataset(fs, baseDir3, properties, new Path("/tmp/src/ds2/daily")); fileStatusList = dataset.getFilesAtPath(fs, baseDir3, pathFilter); Assert.assertEquals(fileStatusList.size(), NUM_LOOKBACK_DAYS + 1); for (FileStatus fileStatus: fileStatusList) { Assert.assertTrue(candidateFiles.contains(PathUtils.getPathWithoutSchemeAndAuthority(fileStatus.getPath()).toString())); } }
@Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { DateTimeFormatter formatter = DateTimeFormat.forPattern(datePattern); LocalDateTime endDate = currentTime; LocalDateTime startDate = endDate.minus(this.lookbackPeriod); DateRangeIterator dateRangeIterator = new DateRangeIterator(startDate, endDate, pattern); List<FileStatus> fileStatuses = Lists.newArrayList(); while (dateRangeIterator.hasNext()) { Path pathWithDateTime = new Path(path, dateRangeIterator.next().toString(formatter)); fileStatuses.addAll(super.getFilesAtPath(fs, pathWithDateTime, fileFilter)); } return fileStatuses; }
TimeAwareRecursiveCopyableDataset extends RecursiveCopyableDataset { @Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { DateTimeFormatter formatter = DateTimeFormat.forPattern(datePattern); LocalDateTime endDate = currentTime; LocalDateTime startDate = endDate.minus(this.lookbackPeriod); DateRangeIterator dateRangeIterator = new DateRangeIterator(startDate, endDate, pattern); List<FileStatus> fileStatuses = Lists.newArrayList(); while (dateRangeIterator.hasNext()) { Path pathWithDateTime = new Path(path, dateRangeIterator.next().toString(formatter)); fileStatuses.addAll(super.getFilesAtPath(fs, pathWithDateTime, fileFilter)); } return fileStatuses; } }
TimeAwareRecursiveCopyableDataset extends RecursiveCopyableDataset { @Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { DateTimeFormatter formatter = DateTimeFormat.forPattern(datePattern); LocalDateTime endDate = currentTime; LocalDateTime startDate = endDate.minus(this.lookbackPeriod); DateRangeIterator dateRangeIterator = new DateRangeIterator(startDate, endDate, pattern); List<FileStatus> fileStatuses = Lists.newArrayList(); while (dateRangeIterator.hasNext()) { Path pathWithDateTime = new Path(path, dateRangeIterator.next().toString(formatter)); fileStatuses.addAll(super.getFilesAtPath(fs, pathWithDateTime, fileFilter)); } return fileStatuses; } TimeAwareRecursiveCopyableDataset(FileSystem fs, Path rootPath, Properties properties, Path glob); }
TimeAwareRecursiveCopyableDataset extends RecursiveCopyableDataset { @Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { DateTimeFormatter formatter = DateTimeFormat.forPattern(datePattern); LocalDateTime endDate = currentTime; LocalDateTime startDate = endDate.minus(this.lookbackPeriod); DateRangeIterator dateRangeIterator = new DateRangeIterator(startDate, endDate, pattern); List<FileStatus> fileStatuses = Lists.newArrayList(); while (dateRangeIterator.hasNext()) { Path pathWithDateTime = new Path(path, dateRangeIterator.next().toString(formatter)); fileStatuses.addAll(super.getFilesAtPath(fs, pathWithDateTime, fileFilter)); } return fileStatuses; } TimeAwareRecursiveCopyableDataset(FileSystem fs, Path rootPath, Properties properties, Path glob);  }
TimeAwareRecursiveCopyableDataset extends RecursiveCopyableDataset { @Override protected List<FileStatus> getFilesAtPath(FileSystem fs, Path path, PathFilter fileFilter) throws IOException { DateTimeFormatter formatter = DateTimeFormat.forPattern(datePattern); LocalDateTime endDate = currentTime; LocalDateTime startDate = endDate.minus(this.lookbackPeriod); DateRangeIterator dateRangeIterator = new DateRangeIterator(startDate, endDate, pattern); List<FileStatus> fileStatuses = Lists.newArrayList(); while (dateRangeIterator.hasNext()) { Path pathWithDateTime = new Path(path, dateRangeIterator.next().toString(formatter)); fileStatuses.addAll(super.getFilesAtPath(fs, pathWithDateTime, fileFilter)); } return fileStatuses; } TimeAwareRecursiveCopyableDataset(FileSystem fs, Path rootPath, Properties properties, Path glob);  static final String DATE_PATTERN_KEY; static final String LOOKBACK_TIME_KEY; static final String DEFAULT_DATE_PATTERN_TIMEZONE; static final String DATE_PATTERN_TIMEZONE_KEY; }