Disable blob level Lineage metrics for FileSystems

apache · Oct 3, 2024 · 9f391a8 · 9f391a8
1 parent 2fb9efc
commit 9f391a8
Show file tree

Hide file tree

Showing 7 changed files with 10 additions and 13 deletions.
diff --git a/...platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystem.java b/...platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/storage/GcsFileSystem.java
@@ -219,7 +219,7 @@ protected String getScheme() {
   protected void reportLineage(GcsResourceId resourceId, Lineage lineage) {
     GcsPath path = resourceId.getGcsPath();
     if (!path.getBucket().isEmpty()) {
-      lineage.add("gcs", ImmutableList.of(path.getBucket(), path.getObject()));
+      lineage.add("gcs", ImmutableList.of(path.getBucket()));
     } else {
       LOG.warn("Report Lineage on relative path {} is unsupported", path.getObject());
     }

diff --git a/...java/io/amazon-web-services/src/main/java/org/apache/beam/sdk/io/aws/s3/S3FileSystem.java b/...java/io/amazon-web-services/src/main/java/org/apache/beam/sdk/io/aws/s3/S3FileSystem.java
@@ -627,7 +627,7 @@ protected S3ResourceId matchNewResource(String singleResourceSpec, boolean isDir
 
   @Override
   protected void reportLineage(S3ResourceId resourceId, Lineage lineage) {
-    lineage.add("s3", ImmutableList.of(resourceId.getBucket(), resourceId.getKey()));
+    lineage.add("s3", ImmutableList.of(resourceId.getBucket()));
   }
 
   /**

diff --git a/...va/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/s3/S3FileSystem.java b/...va/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/s3/S3FileSystem.java
@@ -658,7 +658,7 @@ protected S3ResourceId matchNewResource(String singleResourceSpec, boolean isDir
 
   @Override
   protected void reportLineage(S3ResourceId resourceId, Lineage lineage) {
-    lineage.add("s3", ImmutableList.of(resourceId.getBucket(), resourceId.getKey()));
+    lineage.add("s3", ImmutableList.of(resourceId.getBucket()));
   }
 
   /**

diff --git a/.../azure/src/main/java/org/apache/beam/sdk/io/azure/blobstore/AzureBlobStoreFileSystem.java b/.../azure/src/main/java/org/apache/beam/sdk/io/azure/blobstore/AzureBlobStoreFileSystem.java
@@ -453,13 +453,6 @@ protected AzfsResourceId matchNewResource(String singleResourceSpec, boolean isD
 
   @Override
   protected void reportLineage(AzfsResourceId resourceId, Lineage lineage) {
-    if (!Strings.isNullOrEmpty(resourceId.getBlob())) {
-      lineage.add(
-          "abs",
-          ImmutableList.of(
-              resourceId.getAccount(), resourceId.getContainer(), resourceId.getBlob()));
-    } else {
-      lineage.add("abs", ImmutableList.of(resourceId.getAccount(), resourceId.getContainer()));
-    }
+    lineage.add("abs", ImmutableList.of(resourceId.getAccount(), resourceId.getContainer()));
   }
 }
diff --git a/sdks/python/apache_beam/io/aws/s3filesystem.py b/sdks/python/apache_beam/io/aws/s3filesystem.py
@@ -321,4 +321,6 @@ def report_lineage(self, path, lineage):
     except ValueError:
       # report lineage is fail-safe
       return
+    if len(components) > 1:
+      components = components[:-1]
     lineage.add('s3', *components)
diff --git a/sdks/python/apache_beam/io/azure/blobstoragefilesystem.py b/sdks/python/apache_beam/io/azure/blobstoragefilesystem.py
@@ -323,4 +323,6 @@ def report_lineage(self, path, lineage):
     except ValueError:
       # report lineage is fail-safe
       return
+    if len(components) > 1:
+      components = components[:-1]
     lineage.add('abs', *components)
diff --git a/sdks/python/apache_beam/io/gcp/gcsfilesystem.py b/sdks/python/apache_beam/io/gcp/gcsfilesystem.py
@@ -368,8 +368,8 @@ def delete(self, paths):
 
   def report_lineage(self, path, lineage):
     try:
-      bucket, blob = gcsio.parse_gcs_path(path)
+      bucket, _ = gcsio.parse_gcs_path(path)
     except ValueError:
       # report lineage is fail-safe
       return
-    lineage.add('gcs', bucket, blob)
+    lineage.add('gcs', bucket)