@Test public void testGetLeafDatasetURIs() throws URISyntaxException, IOException { Collection<URI> allDatasetURIs = new ArrayList<URI>(); allDatasetURIs.add(new URI("/data/derived/browsemaps/entities/anet")); allDatasetURIs.add(new URI("/data/derived/browsemaps/entities/comp")); allDatasetURIs.add(new URI("/data/derived/gowl/pymk/invitationsCreationsSends/hourly_data/aggregation/daily")); allDatasetURIs.add(new URI("/data/derived/gowl/pymk/invitationsCreationsSends/hourly_data/aggregation/daily_dedup")); allDatasetURIs.add(new URI("/data/derived")); allDatasetURIs.add(new URI("/data/derived/browsemaps")); allDatasetURIs.add(new URI("/data/derived/browsemaps/entities/")); allDatasetURIs.add(new URI("/data/derived/gowl/")); allDatasetURIs.add(new URI("/data/derived/gowl/pymk/")); allDatasetURIs.add(new URI("/data/derived/gowl/pymk/invitationsCreationsSends/")); allDatasetURIs.add(new URI("/data/derived/gowl/pymk/invitationsCreationsSends/hourly_data/aggregation")); allDatasetURIs.add(new URI("/data/derived2")); Set<URI> disabled = new HashSet<URI>(); disabled.add(new URI("/data/derived/gowl/pymk/invitationsCreationsSends/hourly_data/aggregation/daily")); Set<URI> validURIs = ConfigBasedDatasetsFinder.getValidDatasetURIsHelper(allDatasetURIs, disabled, new Path("/data/derived")); Assert.assertTrue(validURIs.size() == 3); Assert.assertTrue(validURIs.contains(new URI("/data/derived/gowl/pymk/invitationsCreationsSends/hourly_data/aggregation/daily_dedup"))); Assert.assertTrue(validURIs.contains(new URI("/data/derived/browsemaps/entities/comp"))); Assert.assertTrue(validURIs.contains(new URI("/data/derived/browsemaps/entities/anet"))); }
protected static Set<URI> getValidDatasetURIsHelper(Collection<URI> allDatasetURIs, Set<URI> disabledURISet, Path datasetCommonRoot){ if (allDatasetURIs == null || allDatasetURIs.isEmpty()) { return ImmutableSet.of(); } Comparator<URI> pathLengthComparator = new Comparator<URI>() { public int compare(URI c1, URI c2) { return c1.getPath().length() - c2.getPath().length(); } }; List<URI> sortedDatasetsList = new ArrayList<URI>(allDatasetURIs); Collections.sort(sortedDatasetsList, pathLengthComparator); TreeSet<URI> uriSet = new TreeSet<URI>(); Set<URI> noneLeaf = new HashSet<URI>(); for (URI u : sortedDatasetsList) { if (PathUtils.isAncestor(datasetCommonRoot, new Path(u.getPath()))) { URI floor = uriSet.floor(u); if (floor != null && PathUtils.isAncestor(new Path(floor.getPath()), new Path(u.getPath()))) { noneLeaf.add(floor); } uriSet.add(u); } } Set<URI> validURISet = new HashSet<URI>(); for (URI u : uriSet) { if (!noneLeaf.contains(u)) { validURISet.add(u); } } for (URI disable : disabledURISet) { if (validURISet.remove(disable)) { log.info("skip disabled dataset " + disable); } else { log.info("There's no URI " + disable + " available in validURISet."); } } return validURISet; }
ConfigBasedDatasetsFinder implements DatasetsFinder { protected static Set<URI> getValidDatasetURIsHelper(Collection<URI> allDatasetURIs, Set<URI> disabledURISet, Path datasetCommonRoot){ if (allDatasetURIs == null || allDatasetURIs.isEmpty()) { return ImmutableSet.of(); } Comparator<URI> pathLengthComparator = new Comparator<URI>() { public int compare(URI c1, URI c2) { return c1.getPath().length() - c2.getPath().length(); } }; List<URI> sortedDatasetsList = new ArrayList<URI>(allDatasetURIs); Collections.sort(sortedDatasetsList, pathLengthComparator); TreeSet<URI> uriSet = new TreeSet<URI>(); Set<URI> noneLeaf = new HashSet<URI>(); for (URI u : sortedDatasetsList) { if (PathUtils.isAncestor(datasetCommonRoot, new Path(u.getPath()))) { URI floor = uriSet.floor(u); if (floor != null && PathUtils.isAncestor(new Path(floor.getPath()), new Path(u.getPath()))) { noneLeaf.add(floor); } uriSet.add(u); } } Set<URI> validURISet = new HashSet<URI>(); for (URI u : uriSet) { if (!noneLeaf.contains(u)) { validURISet.add(u); } } for (URI disable : disabledURISet) { if (validURISet.remove(disable)) { log.info("skip disabled dataset " + disable); } else { log.info("There's no URI " + disable + " available in validURISet."); } } return validURISet; } }
ConfigBasedDatasetsFinder implements DatasetsFinder { protected static Set<URI> getValidDatasetURIsHelper(Collection<URI> allDatasetURIs, Set<URI> disabledURISet, Path datasetCommonRoot){ if (allDatasetURIs == null || allDatasetURIs.isEmpty()) { return ImmutableSet.of(); } Comparator<URI> pathLengthComparator = new Comparator<URI>() { public int compare(URI c1, URI c2) { return c1.getPath().length() - c2.getPath().length(); } }; List<URI> sortedDatasetsList = new ArrayList<URI>(allDatasetURIs); Collections.sort(sortedDatasetsList, pathLengthComparator); TreeSet<URI> uriSet = new TreeSet<URI>(); Set<URI> noneLeaf = new HashSet<URI>(); for (URI u : sortedDatasetsList) { if (PathUtils.isAncestor(datasetCommonRoot, new Path(u.getPath()))) { URI floor = uriSet.floor(u); if (floor != null && PathUtils.isAncestor(new Path(floor.getPath()), new Path(u.getPath()))) { noneLeaf.add(floor); } uriSet.add(u); } } Set<URI> validURISet = new HashSet<URI>(); for (URI u : uriSet) { if (!noneLeaf.contains(u)) { validURISet.add(u); } } for (URI disable : disabledURISet) { if (validURISet.remove(disable)) { log.info("skip disabled dataset " + disable); } else { log.info("There's no URI " + disable + " available in validURISet."); } } return validURISet; } ConfigBasedDatasetsFinder(FileSystem fs, Properties jobProps); }
ConfigBasedDatasetsFinder implements DatasetsFinder { protected static Set<URI> getValidDatasetURIsHelper(Collection<URI> allDatasetURIs, Set<URI> disabledURISet, Path datasetCommonRoot){ if (allDatasetURIs == null || allDatasetURIs.isEmpty()) { return ImmutableSet.of(); } Comparator<URI> pathLengthComparator = new Comparator<URI>() { public int compare(URI c1, URI c2) { return c1.getPath().length() - c2.getPath().length(); } }; List<URI> sortedDatasetsList = new ArrayList<URI>(allDatasetURIs); Collections.sort(sortedDatasetsList, pathLengthComparator); TreeSet<URI> uriSet = new TreeSet<URI>(); Set<URI> noneLeaf = new HashSet<URI>(); for (URI u : sortedDatasetsList) { if (PathUtils.isAncestor(datasetCommonRoot, new Path(u.getPath()))) { URI floor = uriSet.floor(u); if (floor != null && PathUtils.isAncestor(new Path(floor.getPath()), new Path(u.getPath()))) { noneLeaf.add(floor); } uriSet.add(u); } } Set<URI> validURISet = new HashSet<URI>(); for (URI u : uriSet) { if (!noneLeaf.contains(u)) { validURISet.add(u); } } for (URI disable : disabledURISet) { if (validURISet.remove(disable)) { log.info("skip disabled dataset " + disable); } else { log.info("There's no URI " + disable + " available in validURISet."); } } return validURISet; } ConfigBasedDatasetsFinder(FileSystem fs, Properties jobProps); @Override Path commonDatasetRoot(); @Override List<Dataset> findDatasets(); }
ConfigBasedDatasetsFinder implements DatasetsFinder { protected static Set<URI> getValidDatasetURIsHelper(Collection<URI> allDatasetURIs, Set<URI> disabledURISet, Path datasetCommonRoot){ if (allDatasetURIs == null || allDatasetURIs.isEmpty()) { return ImmutableSet.of(); } Comparator<URI> pathLengthComparator = new Comparator<URI>() { public int compare(URI c1, URI c2) { return c1.getPath().length() - c2.getPath().length(); } }; List<URI> sortedDatasetsList = new ArrayList<URI>(allDatasetURIs); Collections.sort(sortedDatasetsList, pathLengthComparator); TreeSet<URI> uriSet = new TreeSet<URI>(); Set<URI> noneLeaf = new HashSet<URI>(); for (URI u : sortedDatasetsList) { if (PathUtils.isAncestor(datasetCommonRoot, new Path(u.getPath()))) { URI floor = uriSet.floor(u); if (floor != null && PathUtils.isAncestor(new Path(floor.getPath()), new Path(u.getPath()))) { noneLeaf.add(floor); } uriSet.add(u); } } Set<URI> validURISet = new HashSet<URI>(); for (URI u : uriSet) { if (!noneLeaf.contains(u)) { validURISet.add(u); } } for (URI disable : disabledURISet) { if (validURISet.remove(disable)) { log.info("skip disabled dataset " + disable); } else { log.info("There's no URI " + disable + " available in validURISet."); } } return validURISet; } ConfigBasedDatasetsFinder(FileSystem fs, Properties jobProps); @Override Path commonDatasetRoot(); @Override List<Dataset> findDatasets(); static final String GOBBLIN_CONFIG_STORE_WHITELIST_TAG; static final String GOBBLIN_CONFIG_STORE_BLACKLIST_TAGS; static final String GOBBLIN_CONFIG_STORE_DATASET_COMMON_ROOT; static final String JOB_LEVEL_BLACKLIST; static final String WATERMARK_ENABLE; }