Abacn commented on code in PR #32662:
URL: https://github.com/apache/beam/pull/32662#discussion_r1792494730
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSource.java:
##########
@@ -316,18 +318,31 @@ public final List<? extends FileBasedSource<T>> split(
}
}
- /** Report source Lineage. Depend on the number of files, report full file
name or only dir. */
- private void reportSourceLineage(List<Metadata> expandedFiles) {
+ /**
+ * Report source Lineage. Due to the size limit of Beam metrics, report full
file name or only dir
+ * depend on the number of files.
+ *
+ * <p>- Number of files<=100, report full file paths;
+ *
+ * <p>- Number of directory<=100, report up to 100 full directories;
+ *
+ * <p>- Otherwise, report top level only
+ */
+ private static void reportSourceLineage(List<Metadata> expandedFiles) {
if (expandedFiles.size() <= 100) {
for (Metadata metadata : expandedFiles) {
FileSystems.reportSourceLineage(metadata.resourceId());
}
} else {
+ HashSet<ResourceId> hashSet = new HashSet<>();
for (Metadata metadata : expandedFiles) {
- // TODO(yathu) Currently it simply report one level up if num of files
exceeded 100.
- // Consider more dedicated strategy (e.g. resolve common ancestor)
for accurancy, and work
- // with metrics size limit.
-
FileSystems.reportSourceLineage(metadata.resourceId().getCurrentDirectory());
+ ResourceId dir = metadata.resourceId().getCurrentDirectory();
+ hashSet.add(dir);
+ if (hashSet.size() > 100) {
+ FileSystems.reportSourceLineage(dir, LineageLevel.TOP_LEVEL);
+ break;
+ }
+ FileSystems.reportSourceLineage(dir);
Review Comment:
Still working on the branch actively. Now fixed. Will ask for PTAL when test
passes.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
