You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "arpit09 (via GitHub)" <gi...@apache.org> on 2024/02/26 04:06:29 UTC
[PR] Adding an iterator for fetching dataset versions [gobblin]
arpit09 opened a new pull request, #3883:
URL: https://github.com/apache/gobblin/pull/3883
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [x] My PR addresses the following [GOBBLIN-2006](https://issues.apache.org/jira/browse/GOBBLIN-2006) issues and references them in the PR title.
### Description
- [x] Here are some details about my PR, including screenshots (if applicable):
- Add a new method findDatasetVersion in DatasetVersion Interface that returns an iterator
- Implementing an iterator that returns a dataset version that matches the glob pattern passed
- Calling the iterator in the retention job (MultiVersionCleanableDatasetBase.java), made it batch based of 10.
### Tests
- [x] Added the UT for the new iterator based approach
- [x] Updated the UT for the retention job Cleaner
- [x] Retention job integration test are being passed
### Commits
- [x] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1502683173
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/version/finder/AbstractDatasetVersionFinder.java:
##########
@@ -17,21 +17,22 @@
package org.apache.gobblin.data.management.version.finder;
+import com.google.common.collect.Lists;
Review Comment:
gobblin using a different codestyle and import orders, please use this https://gobblin.readthedocs.io/en/latest/developer-guide/CodingStyle/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1502913543
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/version/finder/AbstractDatasetVersionFinder.java:
##########
@@ -86,6 +87,110 @@ public Collection<T> findDatasetVersions(Dataset dataset) throws IOException {
return dataSetVersions;
}
+ /**
+ * Find dataset version in the input {@link org.apache.gobblin.dataset}. Dataset versions are subdirectories of the
+ * input {@link org.apache.gobblin.dataset} representing a single manageable unit in the dataset.
+ *
+ * @param dataset {@link org.apache.gobblin.dataset} to directory containing all versions of a dataset
+ * @return - Returns an iterator for fetching each dataset version found.
+ * @throws IOException
+ */
+ @Override
+ public RemoteIterator<T> findDatasetVersion(Dataset dataset) throws IOException {
+ FileSystemDataset fsDataset = (FileSystemDataset) dataset;
+ Path versionGlobStatus = new Path(fsDataset.datasetRoot(), globVersionPattern());
+ return getDatasetVersionIterator(fsDataset.datasetRoot(), getRegexPattern(versionGlobStatus.toString()));
+ }
+
+ /**
+ * Returns an iterator to fetch the dataset versions for the datasets whose path {@link org.apache.hadoop.fs.Path}
+ * starts with the root and matches the globPattern passed
+ *
+ * @param root - Path of the root from which the Dataset Versions have to be returned
+ * @param pathPattern - Pattern to match the dataset version path
+ * @return - an iterator of matched data versions
+ * @throws IOException
+ */
+ public RemoteIterator<T> getDatasetVersionIterator(Path root, String pathPattern) throws IOException {
+ Stack<RemoteIterator<FileStatus>> iteratorStack = new Stack<>();
+ RemoteIterator<FileStatus> fsIterator = fs.listStatusIterator(root);
+ iteratorStack.push(fsIterator);
+ return new RemoteIterator<T>() {
+ FileStatus nextFileStatus = null;
+ boolean isNextFileStatusProcessed = false;
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (iteratorStack.isEmpty()) {
+ return false;
+ }
+ // No need to process if the next() has not been called
+ if (nextFileStatus != null && !isNextFileStatusProcessed) {
+ return true;
+ }
+ nextFileStatus = fetchNextFileStatus(iteratorStack, pathPattern);
+ isNextFileStatusProcessed = false;
Review Comment:
Is this iterator thread safe?
Just check can you use `Iterators.transform()` from guava here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arpit09 (via GitHub)" <gi...@apache.org>.
arpit09 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1512067188
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/version/finder/AbstractDatasetVersionFinder.java:
##########
@@ -86,6 +87,110 @@ public Collection<T> findDatasetVersions(Dataset dataset) throws IOException {
return dataSetVersions;
}
+ /**
+ * Find dataset version in the input {@link org.apache.gobblin.dataset}. Dataset versions are subdirectories of the
+ * input {@link org.apache.gobblin.dataset} representing a single manageable unit in the dataset.
+ *
+ * @param dataset {@link org.apache.gobblin.dataset} to directory containing all versions of a dataset
+ * @return - Returns an iterator for fetching each dataset version found.
+ * @throws IOException
+ */
+ @Override
+ public RemoteIterator<T> findDatasetVersion(Dataset dataset) throws IOException {
+ FileSystemDataset fsDataset = (FileSystemDataset) dataset;
+ Path versionGlobStatus = new Path(fsDataset.datasetRoot(), globVersionPattern());
+ return getDatasetVersionIterator(fsDataset.datasetRoot(), getRegexPattern(versionGlobStatus.toString()));
+ }
+
+ /**
+ * Returns an iterator to fetch the dataset versions for the datasets whose path {@link org.apache.hadoop.fs.Path}
+ * starts with the root and matches the globPattern passed
+ *
+ * @param root - Path of the root from which the Dataset Versions have to be returned
+ * @param pathPattern - Pattern to match the dataset version path
+ * @return - an iterator of matched data versions
+ * @throws IOException
+ */
+ public RemoteIterator<T> getDatasetVersionIterator(Path root, String pathPattern) throws IOException {
+ Stack<RemoteIterator<FileStatus>> iteratorStack = new Stack<>();
+ RemoteIterator<FileStatus> fsIterator = fs.listStatusIterator(root);
+ iteratorStack.push(fsIterator);
+ return new RemoteIterator<T>() {
+ FileStatus nextFileStatus = null;
+ boolean isNextFileStatusProcessed = false;
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (iteratorStack.isEmpty()) {
+ return false;
+ }
+ // No need to process if the next() has not been called
+ if (nextFileStatus != null && !isNextFileStatusProcessed) {
+ return true;
+ }
+ nextFileStatus = fetchNextFileStatus(iteratorStack, pathPattern);
+ isNextFileStatusProcessed = false;
Review Comment:
I have moved the nextfileStatus and isNextFileStatusProcessed as a pair, to avoid partial updates of these 2 variables. Iterators are usually not thread-safe, the caller should avoid sharing iterator among threads.
`Iterators.transform()` is mostly used when we have a collection and want to iterate over it. In our case, if we don't want to bring in a collection of dataset in-memory. We can still pass this iterator to the transform, is that what you are referring here ? Any advantage we get due to it ?
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/version/finder/AbstractDatasetVersionFinder.java:
##########
@@ -17,21 +17,22 @@
package org.apache.gobblin.data.management.version.finder;
+import com.google.common.collect.Lists;
Review Comment:
Done, thanks for pointing out
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/MultiVersionCleanableDatasetBase.java:
##########
@@ -275,39 +278,48 @@ public void clean() throws IOException {
this.log.info(String.format("Cleaning dataset %s. Using version finder %s and policy %s", this,
versionFinder.getClass().getName(), selectionPolicy));
-
- List<T> versions = Lists.newArrayList(versionFinder.findDatasetVersions(this));
-
- if (versions.isEmpty()) {
- this.log.warn("No dataset version can be found. Ignoring.");
- continue;
- }
-
- Collections.sort(versions, Collections.reverseOrder());
-
- Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
-
- cleanImpl(deletableVersions);
-
- List<DatasetVersion> allVersions = Lists.newArrayList();
- for (T ver : versions) {
- allVersions.add(ver);
- }
- for (RetentionAction retentionAction : versionFinderAndPolicy.getRetentionActions()) {
- try {
- retentionAction.execute(allVersions);
- } catch (Throwable t) {
- atLeastOneFailureSeen = true;
- log.error(String.format("RetentionAction %s failed for dataset %s", retentionAction.getClass().getName(),
- this.datasetRoot()), t);
+ // Avoiding OOM by iterating instead of loading all the datasetVersions in memory
+ RemoteIterator<? extends T> versionRemoteIterator = versionFinder.findDatasetVersion(this);
+ // Cleaning Dataset versions in batch of CLEANABLE_DATASET_BATCH_SIZE to avoid OOM
+ List<T> cleanableVersionsBatch = new ArrayList<>();
+ while (versionRemoteIterator.hasNext()) {
+ T version = versionRemoteIterator.next();
+ cleanableVersionsBatch.add(version);
+ if (cleanableVersionsBatch.size() >= CLEANABLE_DATASET_BATCH_SIZE) {
+ boolean isCleanSuccess = cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
}
}
+ if (!cleanableVersionsBatch.isEmpty()) {
+ boolean isCleanSuccess = cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
+ }
}
if (atLeastOneFailureSeen) {
- throw new RuntimeException(String.format(
- "At least one failure happened while processing %s. Look for previous logs for failures", datasetRoot()));
+ throw new RuntimeException(
+ String.format("At least one failure happened while processing %s. Look for previous logs for failures",
+ datasetRoot()));
+ }
+ }
+
+ private boolean cleanDatasetVersions(List<T> versions, VersionSelectionPolicy<T> selectionPolicy,
+ VersionFinderAndPolicy<T> versionFinderAndPolicy) throws IOException {
+ Collections.sort(versions, Collections.reverseOrder());
+ Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
+ cleanImpl(deletableVersions);
+ List<DatasetVersion> allVersions = Lists.newArrayList(versions);
+ for (RetentionAction retentionAction : versionFinderAndPolicy.getRetentionActions()) {
+ try {
+ retentionAction.execute(allVersions);
+ } catch (Throwable t) {
+ log.error(String.format("RetentionAction %s failed for dataset %s", retentionAction.getClass().getName(),
+ this.datasetRoot()), t);
+ return false;
Review Comment:
Yes, returning false here will only set the flag of atleastOneFailure at line 295. The other dataset retention action will go through irrespective of returning false here.
` atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;`
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/version/finder/AbstractDatasetVersionFinder.java:
##########
@@ -86,6 +87,110 @@ public Collection<T> findDatasetVersions(Dataset dataset) throws IOException {
return dataSetVersions;
}
+ /**
+ * Find dataset version in the input {@link org.apache.gobblin.dataset}. Dataset versions are subdirectories of the
+ * input {@link org.apache.gobblin.dataset} representing a single manageable unit in the dataset.
+ *
+ * @param dataset {@link org.apache.gobblin.dataset} to directory containing all versions of a dataset
+ * @return - Returns an iterator for fetching each dataset version found.
+ * @throws IOException
+ */
+ @Override
+ public RemoteIterator<T> findDatasetVersion(Dataset dataset) throws IOException {
+ FileSystemDataset fsDataset = (FileSystemDataset) dataset;
+ Path versionGlobStatus = new Path(fsDataset.datasetRoot(), globVersionPattern());
+ return getDatasetVersionIterator(fsDataset.datasetRoot(), getRegexPattern(versionGlobStatus.toString()));
+ }
+
+ /**
+ * Returns an iterator to fetch the dataset versions for the datasets whose path {@link org.apache.hadoop.fs.Path}
+ * starts with the root and matches the globPattern passed
+ *
+ * @param root - Path of the root from which the Dataset Versions have to be returned
+ * @param pathPattern - Pattern to match the dataset version path
+ * @return - an iterator of matched data versions
+ * @throws IOException
+ */
+ public RemoteIterator<T> getDatasetVersionIterator(Path root, String pathPattern) throws IOException {
+ Stack<RemoteIterator<FileStatus>> iteratorStack = new Stack<>();
+ RemoteIterator<FileStatus> fsIterator = fs.listStatusIterator(root);
+ iteratorStack.push(fsIterator);
+ return new RemoteIterator<T>() {
+ FileStatus nextFileStatus = null;
+ boolean isNextFileStatusProcessed = false;
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (iteratorStack.isEmpty()) {
+ return false;
+ }
+ // No need to process if the next() has not been called
+ if (nextFileStatus != null && !isNextFileStatusProcessed) {
+ return true;
+ }
+ nextFileStatus = fetchNextFileStatus(iteratorStack, pathPattern);
+ isNextFileStatusProcessed = false;
+ return nextFileStatus != null;
+ }
+
+ @Override
+ public T next() throws IOException {
+ if (nextFileStatus == null || isNextFileStatusProcessed) {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ }
+ T datasetVersion = getDatasetVersion(PathUtils.relativizePath(nextFileStatus.getPath(), root), nextFileStatus);
+ isNextFileStatusProcessed = true;
+ return datasetVersion;
+ }
+ };
+ }
+
+ /**
+ * Helper method to find the next filestatus matching the globPattern.
+ * This uses a stack to keep track of the fileStatusIterator returned at each subpaths
+ *
+ * @param iteratorStack
+ * @param globPattern
+ * @return
+ * @throws IOException
+ */
+ private FileStatus fetchNextFileStatus(Stack<RemoteIterator<FileStatus>> iteratorStack,
+ String globPattern) throws IOException {
+ while (!iteratorStack.isEmpty()) {
+ RemoteIterator<FileStatus> latestfsIterator = iteratorStack.pop();
+ while (latestfsIterator.hasNext()) {
+ FileStatus fileStatus = latestfsIterator.next();
+ if (fileStatus.isDirectory()) {
+ iteratorStack.push(fs.listStatusIterator(fileStatus.getPath()));
Review Comment:
Yes, hdfs file system doesn't provide a method that returns an iterator over a glob pattern. It provides it over a file path, so from the root I keep going to the next folder and keep checking all the filestatus if they match the globPattern
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1502683977
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/version/finder/AbstractDatasetVersionFinder.java:
##########
@@ -17,21 +17,22 @@
package org.apache.gobblin.data.management.version.finder;
+import com.google.common.collect.Lists;
Review Comment:
you need to import codestyle-intellij-gobblin.xml file in your IDE
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1503272965
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/version/finder/AbstractDatasetVersionFinder.java:
##########
@@ -86,6 +87,110 @@ public Collection<T> findDatasetVersions(Dataset dataset) throws IOException {
return dataSetVersions;
}
+ /**
+ * Find dataset version in the input {@link org.apache.gobblin.dataset}. Dataset versions are subdirectories of the
+ * input {@link org.apache.gobblin.dataset} representing a single manageable unit in the dataset.
+ *
+ * @param dataset {@link org.apache.gobblin.dataset} to directory containing all versions of a dataset
+ * @return - Returns an iterator for fetching each dataset version found.
+ * @throws IOException
+ */
+ @Override
+ public RemoteIterator<T> findDatasetVersion(Dataset dataset) throws IOException {
+ FileSystemDataset fsDataset = (FileSystemDataset) dataset;
+ Path versionGlobStatus = new Path(fsDataset.datasetRoot(), globVersionPattern());
+ return getDatasetVersionIterator(fsDataset.datasetRoot(), getRegexPattern(versionGlobStatus.toString()));
+ }
+
+ /**
+ * Returns an iterator to fetch the dataset versions for the datasets whose path {@link org.apache.hadoop.fs.Path}
+ * starts with the root and matches the globPattern passed
+ *
+ * @param root - Path of the root from which the Dataset Versions have to be returned
+ * @param pathPattern - Pattern to match the dataset version path
+ * @return - an iterator of matched data versions
+ * @throws IOException
+ */
+ public RemoteIterator<T> getDatasetVersionIterator(Path root, String pathPattern) throws IOException {
+ Stack<RemoteIterator<FileStatus>> iteratorStack = new Stack<>();
+ RemoteIterator<FileStatus> fsIterator = fs.listStatusIterator(root);
+ iteratorStack.push(fsIterator);
+ return new RemoteIterator<T>() {
+ FileStatus nextFileStatus = null;
+ boolean isNextFileStatusProcessed = false;
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (iteratorStack.isEmpty()) {
+ return false;
+ }
+ // No need to process if the next() has not been called
+ if (nextFileStatus != null && !isNextFileStatusProcessed) {
+ return true;
+ }
+ nextFileStatus = fetchNextFileStatus(iteratorStack, pathPattern);
+ isNextFileStatusProcessed = false;
+ return nextFileStatus != null;
+ }
+
+ @Override
+ public T next() throws IOException {
+ if (nextFileStatus == null || isNextFileStatusProcessed) {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ }
+ T datasetVersion = getDatasetVersion(PathUtils.relativizePath(nextFileStatus.getPath(), root), nextFileStatus);
+ isNextFileStatusProcessed = true;
+ return datasetVersion;
+ }
+ };
+ }
+
+ /**
+ * Helper method to find the next filestatus matching the globPattern.
+ * This uses a stack to keep track of the fileStatusIterator returned at each subpaths
+ *
+ * @param iteratorStack
+ * @param globPattern
+ * @return
+ * @throws IOException
+ */
+ private FileStatus fetchNextFileStatus(Stack<RemoteIterator<FileStatus>> iteratorStack,
+ String globPattern) throws IOException {
+ while (!iteratorStack.isEmpty()) {
+ RemoteIterator<FileStatus> latestfsIterator = iteratorStack.pop();
+ while (latestfsIterator.hasNext()) {
+ FileStatus fileStatus = latestfsIterator.next();
+ if (fileStatus.isDirectory()) {
+ iteratorStack.push(fs.listStatusIterator(fileStatus.getPath()));
Review Comment:
You are doing it this way because `listStatusIterator` does not accept a glob pattern?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1512185406
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/MultiVersionCleanableDatasetBase.java:
##########
@@ -275,42 +282,55 @@ public void clean() throws IOException {
this.log.info(String.format("Cleaning dataset %s. Using version finder %s and policy %s", this,
versionFinder.getClass().getName(), selectionPolicy));
-
- List<T> versions = Lists.newArrayList(versionFinder.findDatasetVersions(this));
-
- if (versions.isEmpty()) {
- this.log.warn("No dataset version can be found. Ignoring.");
- continue;
- }
-
- Collections.sort(versions, Collections.reverseOrder());
-
- Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
-
- cleanImpl(deletableVersions);
-
- List<DatasetVersion> allVersions = Lists.newArrayList();
- for (T ver : versions) {
- allVersions.add(ver);
- }
- for (RetentionAction retentionAction : versionFinderAndPolicy.getRetentionActions()) {
- try {
- retentionAction.execute(allVersions);
- } catch (Throwable t) {
- atLeastOneFailureSeen = true;
- log.error(String.format("RetentionAction %s failed for dataset %s", retentionAction.getClass().getName(),
- this.datasetRoot()), t);
+ // Avoiding OOM by iterating instead of loading all the datasetVersions in memory
+ RemoteIterator<? extends T> versionRemoteIterator = versionFinder.findDatasetVersion(this);
+ // Cleaning Dataset versions in batch of CLEANABLE_DATASET_BATCH_SIZE to avoid OOM
+ List<T> cleanableVersionsBatch = new ArrayList<>();
+ while (versionRemoteIterator.hasNext()) {
+ T version = versionRemoteIterator.next();
+ cleanableVersionsBatch.add(version);
+ if (cleanableVersionsBatch.size() >= CLEANABLE_DATASET_BATCH_SIZE) {
+ boolean isCleanSuccess =
+ cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
}
}
+ if (!cleanableVersionsBatch.isEmpty()) {
+ boolean isCleanSuccess = cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
+ }
}
if (atLeastOneFailureSeen) {
- throw new RuntimeException(String.format(
- "At least one failure happened while processing %s. Look for previous logs for failures", datasetRoot()));
+ throw new RuntimeException(
+ String.format("At least one failure happened while processing %s. Look for previous logs for failures",
+ datasetRoot()));
}
}
- protected void cleanImpl(Collection<T> deletableVersions) throws IOException {
+ private boolean cleanDatasetVersions(List<T> versions, VersionSelectionPolicy<T> selectionPolicy,
+ VersionFinderAndPolicy<T> versionFinderAndPolicy)
+ throws IOException {
+ boolean isCleanSuccess = true;
+ Collections.sort(versions, Collections.reverseOrder());
+ Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
Review Comment:
Sorting a subset of datasets and then calling `listSelectedVersions` on it should be okay in most of the selection policies. But for NewestKSelectionPolicy, we need to supply it with complete list of sorted datasets to find absolute K newest datasets.
Second problem with using this logic with NewestKSelectionPolicy is that it may delete K datasets multiple times, effectively cleaning m x K datasets.
What is your thought on tackling this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1520941490
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/policy/NewestKSelectionPolicy.java:
##########
@@ -61,6 +63,12 @@ public class NewestKSelectionPolicy<T extends DatasetVersion> implements Version
*/
public static final String NEWEST_K_VERSIONS_NOTSELECTED_KEY = "selection.newestK.versionsNotSelected";
+ /**
+ * This denotes to use iterator for fetching all the versions of dataset.
Review Comment:
Please also add why one should consider setting this config
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arpit09 (via GitHub)" <gi...@apache.org>.
arpit09 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1520991828
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/policy/NewestKSelectionPolicy.java:
##########
@@ -61,6 +63,12 @@ public class NewestKSelectionPolicy<T extends DatasetVersion> implements Version
*/
public static final String NEWEST_K_VERSIONS_NOTSELECTED_KEY = "selection.newestK.versionsNotSelected";
+ /**
+ * This denotes to use iterator for fetching all the versions of dataset.
+ * If true, this can't be used along with NewestKSelectionPolicy
+ */
+ public static final String VERSION_FINDER_ITERATOR = "version.finder.iterator";
Review Comment:
Done
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/policy/NewestKSelectionPolicy.java:
##########
@@ -61,6 +63,12 @@ public class NewestKSelectionPolicy<T extends DatasetVersion> implements Version
*/
public static final String NEWEST_K_VERSIONS_NOTSELECTED_KEY = "selection.newestK.versionsNotSelected";
+ /**
+ * This denotes to use iterator for fetching all the versions of dataset.
Review Comment:
Added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1512185917
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/MultiVersionCleanableDatasetBase.java:
##########
@@ -275,42 +282,55 @@ public void clean() throws IOException {
this.log.info(String.format("Cleaning dataset %s. Using version finder %s and policy %s", this,
versionFinder.getClass().getName(), selectionPolicy));
-
- List<T> versions = Lists.newArrayList(versionFinder.findDatasetVersions(this));
-
- if (versions.isEmpty()) {
- this.log.warn("No dataset version can be found. Ignoring.");
- continue;
- }
-
- Collections.sort(versions, Collections.reverseOrder());
-
- Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
-
- cleanImpl(deletableVersions);
-
- List<DatasetVersion> allVersions = Lists.newArrayList();
- for (T ver : versions) {
- allVersions.add(ver);
- }
- for (RetentionAction retentionAction : versionFinderAndPolicy.getRetentionActions()) {
- try {
- retentionAction.execute(allVersions);
- } catch (Throwable t) {
- atLeastOneFailureSeen = true;
- log.error(String.format("RetentionAction %s failed for dataset %s", retentionAction.getClass().getName(),
- this.datasetRoot()), t);
+ // Avoiding OOM by iterating instead of loading all the datasetVersions in memory
+ RemoteIterator<? extends T> versionRemoteIterator = versionFinder.findDatasetVersion(this);
+ // Cleaning Dataset versions in batch of CLEANABLE_DATASET_BATCH_SIZE to avoid OOM
+ List<T> cleanableVersionsBatch = new ArrayList<>();
+ while (versionRemoteIterator.hasNext()) {
+ T version = versionRemoteIterator.next();
+ cleanableVersionsBatch.add(version);
+ if (cleanableVersionsBatch.size() >= CLEANABLE_DATASET_BATCH_SIZE) {
+ boolean isCleanSuccess =
+ cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
}
}
+ if (!cleanableVersionsBatch.isEmpty()) {
+ boolean isCleanSuccess = cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
+ }
}
if (atLeastOneFailureSeen) {
- throw new RuntimeException(String.format(
- "At least one failure happened while processing %s. Look for previous logs for failures", datasetRoot()));
+ throw new RuntimeException(
+ String.format("At least one failure happened while processing %s. Look for previous logs for failures",
+ datasetRoot()));
}
}
- protected void cleanImpl(Collection<T> deletableVersions) throws IOException {
+ private boolean cleanDatasetVersions(List<T> versions, VersionSelectionPolicy<T> selectionPolicy,
+ VersionFinderAndPolicy<T> versionFinderAndPolicy)
+ throws IOException {
+ boolean isCleanSuccess = true;
+ Collections.sort(versions, Collections.reverseOrder());
+ Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
Review Comment:
What was the selection policy where this issue of OOM was detected?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arpit09 (via GitHub)" <gi...@apache.org>.
arpit09 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1520898002
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/MultiVersionCleanableDatasetBase.java:
##########
@@ -275,42 +282,55 @@ public void clean() throws IOException {
this.log.info(String.format("Cleaning dataset %s. Using version finder %s and policy %s", this,
versionFinder.getClass().getName(), selectionPolicy));
-
- List<T> versions = Lists.newArrayList(versionFinder.findDatasetVersions(this));
-
- if (versions.isEmpty()) {
- this.log.warn("No dataset version can be found. Ignoring.");
- continue;
- }
-
- Collections.sort(versions, Collections.reverseOrder());
-
- Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
-
- cleanImpl(deletableVersions);
-
- List<DatasetVersion> allVersions = Lists.newArrayList();
- for (T ver : versions) {
- allVersions.add(ver);
- }
- for (RetentionAction retentionAction : versionFinderAndPolicy.getRetentionActions()) {
- try {
- retentionAction.execute(allVersions);
- } catch (Throwable t) {
- atLeastOneFailureSeen = true;
- log.error(String.format("RetentionAction %s failed for dataset %s", retentionAction.getClass().getName(),
- this.datasetRoot()), t);
+ // Avoiding OOM by iterating instead of loading all the datasetVersions in memory
+ RemoteIterator<? extends T> versionRemoteIterator = versionFinder.findDatasetVersion(this);
+ // Cleaning Dataset versions in batch of CLEANABLE_DATASET_BATCH_SIZE to avoid OOM
+ List<T> cleanableVersionsBatch = new ArrayList<>();
+ while (versionRemoteIterator.hasNext()) {
+ T version = versionRemoteIterator.next();
+ cleanableVersionsBatch.add(version);
+ if (cleanableVersionsBatch.size() >= CLEANABLE_DATASET_BATCH_SIZE) {
+ boolean isCleanSuccess =
+ cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
}
}
+ if (!cleanableVersionsBatch.isEmpty()) {
+ boolean isCleanSuccess = cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
+ }
}
if (atLeastOneFailureSeen) {
- throw new RuntimeException(String.format(
- "At least one failure happened while processing %s. Look for previous logs for failures", datasetRoot()));
+ throw new RuntimeException(
+ String.format("At least one failure happened while processing %s. Look for previous logs for failures",
+ datasetRoot()));
}
}
- protected void cleanImpl(Collection<T> deletableVersions) throws IOException {
+ private boolean cleanDatasetVersions(List<T> versions, VersionSelectionPolicy<T> selectionPolicy,
+ VersionFinderAndPolicy<T> versionFinderAndPolicy)
+ throws IOException {
+ boolean isCleanSuccess = true;
+ Collections.sort(versions, Collections.reverseOrder());
+ Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
Review Comment:
Move to a config driven iterator based approach,
When `gobblin.retention.version.finder.iterator` is set to `true` then only iterator base approach will be used, if not default list based approach will be used.
`gobblin.retention.version.finder.iterator` can't be set true along with NewestKSelectionPolicy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1520941213
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/policy/NewestKSelectionPolicy.java:
##########
@@ -61,6 +63,12 @@ public class NewestKSelectionPolicy<T extends DatasetVersion> implements Version
*/
public static final String NEWEST_K_VERSIONS_NOTSELECTED_KEY = "selection.newestK.versionsNotSelected";
+ /**
+ * This denotes to use iterator for fetching all the versions of dataset.
+ * If true, this can't be used along with NewestKSelectionPolicy
+ */
+ public static final String VERSION_FINDER_ITERATOR = "version.finder.iterator";
Review Comment:
lets name it interrogative type , how about SHOULD_ITERATE_VERSIONS ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2006] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 merged PR #3883:
URL: https://github.com/apache/gobblin/pull/3883
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#issuecomment-1963434433
## [Codecov](https://app.codecov.io/gh/apache/gobblin/pull/3883?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 41.82%. Comparing base [(`436cf26`)](https://app.codecov.io/gh/apache/gobblin/commit/436cf26cc23ae189936495d7c63a06c77395ca62?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`9cc8586`)](https://app.codecov.io/gh/apache/gobblin/pull/3883?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 1 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #3883 +/- ##
============================================
- Coverage 46.73% 41.82% -4.92%
+ Complexity 11150 2197 -8953
============================================
Files 2216 469 -1747
Lines 87523 19910 -67613
Branches 9616 2321 -7295
============================================
- Hits 40905 8327 -32578
+ Misses 42925 10700 -32225
+ Partials 3693 883 -2810
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/gobblin/pull/3883?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1502673092
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/MultiVersionCleanableDatasetBase.java:
##########
@@ -275,39 +278,48 @@ public void clean() throws IOException {
this.log.info(String.format("Cleaning dataset %s. Using version finder %s and policy %s", this,
versionFinder.getClass().getName(), selectionPolicy));
-
- List<T> versions = Lists.newArrayList(versionFinder.findDatasetVersions(this));
-
- if (versions.isEmpty()) {
- this.log.warn("No dataset version can be found. Ignoring.");
- continue;
- }
-
- Collections.sort(versions, Collections.reverseOrder());
-
- Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
-
- cleanImpl(deletableVersions);
-
- List<DatasetVersion> allVersions = Lists.newArrayList();
- for (T ver : versions) {
- allVersions.add(ver);
- }
- for (RetentionAction retentionAction : versionFinderAndPolicy.getRetentionActions()) {
- try {
- retentionAction.execute(allVersions);
- } catch (Throwable t) {
- atLeastOneFailureSeen = true;
- log.error(String.format("RetentionAction %s failed for dataset %s", retentionAction.getClass().getName(),
- this.datasetRoot()), t);
+ // Avoiding OOM by iterating instead of loading all the datasetVersions in memory
+ RemoteIterator<? extends T> versionRemoteIterator = versionFinder.findDatasetVersion(this);
+ // Cleaning Dataset versions in batch of CLEANABLE_DATASET_BATCH_SIZE to avoid OOM
+ List<T> cleanableVersionsBatch = new ArrayList<>();
+ while (versionRemoteIterator.hasNext()) {
+ T version = versionRemoteIterator.next();
+ cleanableVersionsBatch.add(version);
+ if (cleanableVersionsBatch.size() >= CLEANABLE_DATASET_BATCH_SIZE) {
+ boolean isCleanSuccess = cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
}
}
+ if (!cleanableVersionsBatch.isEmpty()) {
+ boolean isCleanSuccess = cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
+ }
}
if (atLeastOneFailureSeen) {
- throw new RuntimeException(String.format(
- "At least one failure happened while processing %s. Look for previous logs for failures", datasetRoot()));
+ throw new RuntimeException(
+ String.format("At least one failure happened while processing %s. Look for previous logs for failures",
+ datasetRoot()));
+ }
+ }
+
+ private boolean cleanDatasetVersions(List<T> versions, VersionSelectionPolicy<T> selectionPolicy,
+ VersionFinderAndPolicy<T> versionFinderAndPolicy) throws IOException {
+ Collections.sort(versions, Collections.reverseOrder());
+ Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
+ cleanImpl(deletableVersions);
+ List<DatasetVersion> allVersions = Lists.newArrayList(versions);
+ for (RetentionAction retentionAction : versionFinderAndPolicy.getRetentionActions()) {
+ try {
+ retentionAction.execute(allVersions);
+ } catch (Throwable t) {
+ log.error(String.format("RetentionAction %s failed for dataset %s", retentionAction.getClass().getName(),
+ this.datasetRoot()), t);
+ return false;
Review Comment:
We still need to try other retention actions on this dataset and also remove this version from `versions` before returning, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adding an iterator for fetching dataset versions [gobblin]
Posted by "arpit09 (via GitHub)" <gi...@apache.org>.
arpit09 commented on code in PR #3883:
URL: https://github.com/apache/gobblin/pull/3883#discussion_r1512388680
##########
gobblin-data-management/src/main/java/org/apache/gobblin/data/management/retention/dataset/MultiVersionCleanableDatasetBase.java:
##########
@@ -275,42 +282,55 @@ public void clean() throws IOException {
this.log.info(String.format("Cleaning dataset %s. Using version finder %s and policy %s", this,
versionFinder.getClass().getName(), selectionPolicy));
-
- List<T> versions = Lists.newArrayList(versionFinder.findDatasetVersions(this));
-
- if (versions.isEmpty()) {
- this.log.warn("No dataset version can be found. Ignoring.");
- continue;
- }
-
- Collections.sort(versions, Collections.reverseOrder());
-
- Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
-
- cleanImpl(deletableVersions);
-
- List<DatasetVersion> allVersions = Lists.newArrayList();
- for (T ver : versions) {
- allVersions.add(ver);
- }
- for (RetentionAction retentionAction : versionFinderAndPolicy.getRetentionActions()) {
- try {
- retentionAction.execute(allVersions);
- } catch (Throwable t) {
- atLeastOneFailureSeen = true;
- log.error(String.format("RetentionAction %s failed for dataset %s", retentionAction.getClass().getName(),
- this.datasetRoot()), t);
+ // Avoiding OOM by iterating instead of loading all the datasetVersions in memory
+ RemoteIterator<? extends T> versionRemoteIterator = versionFinder.findDatasetVersion(this);
+ // Cleaning Dataset versions in batch of CLEANABLE_DATASET_BATCH_SIZE to avoid OOM
+ List<T> cleanableVersionsBatch = new ArrayList<>();
+ while (versionRemoteIterator.hasNext()) {
+ T version = versionRemoteIterator.next();
+ cleanableVersionsBatch.add(version);
+ if (cleanableVersionsBatch.size() >= CLEANABLE_DATASET_BATCH_SIZE) {
+ boolean isCleanSuccess =
+ cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
}
}
+ if (!cleanableVersionsBatch.isEmpty()) {
+ boolean isCleanSuccess = cleanDatasetVersions(cleanableVersionsBatch, selectionPolicy, versionFinderAndPolicy);
+ atLeastOneFailureSeen = atLeastOneFailureSeen || !isCleanSuccess;
+ }
}
if (atLeastOneFailureSeen) {
- throw new RuntimeException(String.format(
- "At least one failure happened while processing %s. Look for previous logs for failures", datasetRoot()));
+ throw new RuntimeException(
+ String.format("At least one failure happened while processing %s. Look for previous logs for failures",
+ datasetRoot()));
}
}
- protected void cleanImpl(Collection<T> deletableVersions) throws IOException {
+ private boolean cleanDatasetVersions(List<T> versions, VersionSelectionPolicy<T> selectionPolicy,
+ VersionFinderAndPolicy<T> versionFinderAndPolicy)
+ throws IOException {
+ boolean isCleanSuccess = true;
+ Collections.sort(versions, Collections.reverseOrder());
+ Collection<T> deletableVersions = selectionPolicy.listSelectedVersions(versions);
Review Comment:
For OOM, The selection policy excluded 2 items that was in combination with SelectTimeBeforePolicy (time < 3 days old).
For the Iterator-based approach, NewestKselectionpolicy doesn't add any value as it needs all the data in memory which we want to avoid in our case. We can add a config, allowing the client to go via an iterator-based approach or list-based. If the iterator is chosen with NewestKSelectionPolicy - we can throw a runtime error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org