This is an automated email from the ASF dual-hosted git repository.
yhu pushed a commit to branch release-2.63
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/release-2.63 by this push:
new e285617d304 Revert emitting lineage as boundedtrie from python (#33935)
e285617d304 is described below
commit e285617d30432ab3e13c29a152c6ffbe62672bee
Author: Rohit <[email protected]>
AuthorDate: Tue Feb 11 07:30:30 2025 -0800
Revert emitting lineage as boundedtrie from python (#33935)
* Revert "Place bounded trie lineage metrics under a new name."
This reverts commit 3210b47d94ea9cd032eb6b75904fb25b3da47dd4.
* Revert "Merge pull request #33381 Migrate lineage counters to bounded
tries."
This reverts commit e8df26fcea46e3ca38d02ac1263389dbf859472c, reversing
changes made to 8fee3ca5753ed2ca7563e46ce17cef4d9d95e3cc.
---
sdks/python/apache_beam/io/aws/s3filesystem.py | 10 ++--
.../python/apache_beam/io/aws/s3filesystem_test.py | 3 +-
.../apache_beam/io/azure/blobstoragefilesystem.py | 10 ++--
.../io/azure/blobstoragefilesystem_test.py | 3 +-
sdks/python/apache_beam/io/filebasedsink.py | 20 +++++--
sdks/python/apache_beam/io/filebasedsource.py | 53 ++++++++++++++++-
sdks/python/apache_beam/io/filesystem.py | 6 +-
sdks/python/apache_beam/io/filesystems.py | 16 ++++--
sdks/python/apache_beam/io/gcp/bigquery.py | 2 +-
sdks/python/apache_beam/io/gcp/gcsfilesystem.py | 10 ++--
.../apache_beam/io/gcp/gcsfilesystem_test.py | 3 +-
sdks/python/apache_beam/io/localfilesystem.py | 3 -
sdks/python/apache_beam/metrics/cells.py | 7 +--
sdks/python/apache_beam/metrics/metric.py | 67 +++++-----------------
sdks/python/apache_beam/metrics/metric_test.py | 11 +---
15 files changed, 118 insertions(+), 106 deletions(-)
diff --git a/sdks/python/apache_beam/io/aws/s3filesystem.py
b/sdks/python/apache_beam/io/aws/s3filesystem.py
index 584263ec241..ffbce5893a9 100644
--- a/sdks/python/apache_beam/io/aws/s3filesystem.py
+++ b/sdks/python/apache_beam/io/aws/s3filesystem.py
@@ -18,7 +18,6 @@
"""S3 file system implementation for accessing files on AWS S3."""
# pytype: skip-file
-import traceback
from apache_beam.io.aws import s3io
from apache_beam.io.filesystem import BeamIOError
@@ -316,13 +315,14 @@ class S3FileSystem(FileSystem):
if exceptions:
raise BeamIOError("Delete operation failed", exceptions)
- def report_lineage(self, path, lineage):
+ def report_lineage(self, path, lineage, level=None):
try:
components = s3io.parse_s3_path(path, object_optional=True)
except ValueError:
# report lineage is fail-safe
- traceback.print_exc()
return
- if components and not components[-1]:
+ if level == FileSystem.LineageLevel.TOP_LEVEL or \
+ (len(components) > 1 and components[-1] == ''):
+ # bucket only
components = components[:-1]
- lineage.add('s3', *components, last_segment_sep='/')
+ lineage.add('s3', *components)
diff --git a/sdks/python/apache_beam/io/aws/s3filesystem_test.py
b/sdks/python/apache_beam/io/aws/s3filesystem_test.py
index 036727cd7a7..87403f482bd 100644
--- a/sdks/python/apache_beam/io/aws/s3filesystem_test.py
+++ b/sdks/python/apache_beam/io/aws/s3filesystem_test.py
@@ -272,8 +272,7 @@ class S3FileSystemTest(unittest.TestCase):
def _verify_lineage(self, uri, expected_segments):
lineage_mock = mock.MagicMock()
self.fs.report_lineage(uri, lineage_mock)
- lineage_mock.add.assert_called_once_with(
- "s3", *expected_segments, last_segment_sep='/')
+ lineage_mock.add.assert_called_once_with("s3", *expected_segments)
if __name__ == '__main__':
diff --git a/sdks/python/apache_beam/io/azure/blobstoragefilesystem.py
b/sdks/python/apache_beam/io/azure/blobstoragefilesystem.py
index 4b7462cae03..4495245dc54 100644
--- a/sdks/python/apache_beam/io/azure/blobstoragefilesystem.py
+++ b/sdks/python/apache_beam/io/azure/blobstoragefilesystem.py
@@ -18,7 +18,6 @@
"""Azure Blob Storage Implementation for accesing files on
Azure Blob Storage.
"""
-import traceback
from apache_beam.io.azure import blobstorageio
from apache_beam.io.filesystem import BeamIOError
@@ -318,14 +317,15 @@ class BlobStorageFileSystem(FileSystem):
if exceptions:
raise BeamIOError("Delete operation failed", exceptions)
- def report_lineage(self, path, lineage):
+ def report_lineage(self, path, lineage, level=None):
try:
components = blobstorageio.parse_azfs_path(
path, blob_optional=True, get_account=True)
except ValueError:
# report lineage is fail-safe
- traceback.print_exc()
return
- if components and not components[-1]:
+ if level == FileSystem.LineageLevel.TOP_LEVEL \
+ or(len(components) > 1 and components[-1] == ''):
+ # bucket only
components = components[:-1]
- lineage.add('abs', *components, last_segment_sep='/')
+ lineage.add('abs', *components)
diff --git a/sdks/python/apache_beam/io/azure/blobstoragefilesystem_test.py
b/sdks/python/apache_beam/io/azure/blobstoragefilesystem_test.py
index c3418e137e8..138fe5f78b2 100644
--- a/sdks/python/apache_beam/io/azure/blobstoragefilesystem_test.py
+++ b/sdks/python/apache_beam/io/azure/blobstoragefilesystem_test.py
@@ -330,8 +330,7 @@ class BlobStorageFileSystemTest(unittest.TestCase):
def _verify_lineage(self, uri, expected_segments):
lineage_mock = mock.MagicMock()
self.fs.report_lineage(uri, lineage_mock)
- lineage_mock.add.assert_called_once_with(
- "abs", *expected_segments, last_segment_sep='/')
+ lineage_mock.add.assert_called_once_with("abs", *expected_segments)
if __name__ == '__main__':
diff --git a/sdks/python/apache_beam/io/filebasedsink.py
b/sdks/python/apache_beam/io/filebasedsink.py
index eb433bd6058..f9d4303c8c7 100644
--- a/sdks/python/apache_beam/io/filebasedsink.py
+++ b/sdks/python/apache_beam/io/filebasedsink.py
@@ -286,16 +286,24 @@ class FileBasedSink(iobase.Sink):
def _report_sink_lineage(self, dst_glob, dst_files):
"""
- Report sink Lineage. Report every file if number of files no more than 10,
- otherwise only report glob.
+ Report sink Lineage. Report every file if number of files no more than 100,
+ otherwise only report at directory level.
"""
- # There is rollup at the higher level, but this loses glob information.
- # Better to report multiple globs than just the parent directory.
- if len(dst_files) <= 10:
+ if len(dst_files) <= 100:
for dst in dst_files:
FileSystems.report_sink_lineage(dst)
else:
- FileSystems.report_sink_lineage(dst_glob)
+ dst = dst_glob
+ # dst_glob has a wildcard for shard number (see _shard_name_template)
+ sep = dst_glob.find('*')
+ if sep > 0:
+ dst = dst[:sep]
+ try:
+ dst, _ = FileSystems.split(dst)
+ except ValueError:
+ return # lineage report is fail-safe
+
+ FileSystems.report_sink_lineage(dst)
@check_accessible(['file_path_prefix'])
def finalize_write(
diff --git a/sdks/python/apache_beam/io/filebasedsource.py
b/sdks/python/apache_beam/io/filebasedsource.py
index 49b1b1d125f..a02bc6de32c 100644
--- a/sdks/python/apache_beam/io/filebasedsource.py
+++ b/sdks/python/apache_beam/io/filebasedsource.py
@@ -39,6 +39,7 @@ from apache_beam.io import iobase
from apache_beam.io import range_trackers
from apache_beam.io.filesystem import CompressionTypes
from apache_beam.io.filesystem import FileMetadata
+from apache_beam.io.filesystem import FileSystem
from apache_beam.io.filesystems import FileSystems
from apache_beam.io.restriction_trackers import OffsetRange
from apache_beam.options.value_provider import StaticValueProvider
@@ -169,11 +170,37 @@ class FileBasedSource(iobase.BoundedSource):
splittable=splittable)
single_file_sources.append(single_file_source)
- FileSystems.report_source_lineage(pattern)
+ self._report_source_lineage(files_metadata)
self._concat_source = concat_source.ConcatSource(single_file_sources)
return self._concat_source
+ def _report_source_lineage(self, files_metadata):
+ """
+ Report source Lineage. depend on the number of files, report full file
+ name, only dir, or only top level
+ """
+ if len(files_metadata) <= 100:
+ for file_metadata in files_metadata:
+ FileSystems.report_source_lineage(file_metadata.path)
+ else:
+ size_track = set()
+ for file_metadata in files_metadata:
+ if len(size_track) >= 100:
+ FileSystems.report_source_lineage(
+ file_metadata.path, level=FileSystem.LineageLevel.TOP_LEVEL)
+ return
+
+ try:
+ base, _ = FileSystems.split(file_metadata.path)
+ except ValueError:
+ pass
+ else:
+ size_track.add(base)
+
+ for base in size_track:
+ FileSystems.report_source_lineage(base)
+
def open_file(self, file_name):
return FileSystems.open(
file_name,
@@ -355,7 +382,7 @@ class _ExpandIntoRanges(DoFn):
match_results = FileSystems.match([element])
metadata_list = match_results[0].metadata_list
for metadata in metadata_list:
- FileSystems.report_source_lineage(metadata.path)
+ self._report_source_lineage(metadata.path)
splittable = (
self._splittable and _determine_splittability_from_compression_type(
@@ -370,6 +397,28 @@ class _ExpandIntoRanges(DoFn):
metadata,
OffsetRange(0, range_trackers.OffsetRangeTracker.OFFSET_INFINITY))
+ def _report_source_lineage(self, path):
+ """
+ Report source Lineage. Due to the size limit of Beam metrics, report full
+ file name or only top level depend on the number of files.
+
+ * Number of files<=100, report full file paths;
+
+ * Otherwise, report top level only.
+ """
+ if self._size_track is None:
+ self._size_track = set()
+ elif len(self._size_track) == 0:
+ FileSystems.report_source_lineage(
+ path, level=FileSystem.LineageLevel.TOP_LEVEL)
+ return
+
+ self._size_track.add(path)
+ FileSystems.report_source_lineage(path)
+
+ if len(self._size_track) >= 100:
+ self._size_track.clear()
+
class _ReadRange(DoFn):
def __init__(
diff --git a/sdks/python/apache_beam/io/filesystem.py
b/sdks/python/apache_beam/io/filesystem.py
index bdc25dcf0fe..840fdf3309e 100644
--- a/sdks/python/apache_beam/io/filesystem.py
+++ b/sdks/python/apache_beam/io/filesystem.py
@@ -934,7 +934,11 @@ class FileSystem(BeamPlugin, metaclass=abc.ABCMeta):
"""
raise NotImplementedError
- def report_lineage(self, path, unused_lineage):
+ class LineageLevel:
+ FILE = 'FILE'
+ TOP_LEVEL = 'TOP_LEVEL'
+
+ def report_lineage(self, path, unused_lineage, level=None):
"""
Report Lineage metrics for path.
diff --git a/sdks/python/apache_beam/io/filesystems.py
b/sdks/python/apache_beam/io/filesystems.py
index 1d64f88684b..87f45f3308e 100644
--- a/sdks/python/apache_beam/io/filesystems.py
+++ b/sdks/python/apache_beam/io/filesystems.py
@@ -391,21 +391,27 @@ class FileSystems(object):
return filesystem.CHUNK_SIZE
@staticmethod
- def report_source_lineage(path):
+ def report_source_lineage(path, level=None):
"""
- Report source :class:`~apache_beam.metrics.metric.Lineage`.
+ Report source :class:`~apache_beam.metrics.metric.LineageLevel`.
Args:
path: string path to be reported.
+ level: the level of file path. default to
+ :class:`~apache_beam.io.filesystem.FileSystem.LineageLevel`.FILE.
"""
- FileSystems.get_filesystem(path).report_lineage(path, Lineage.sources())
+ filesystem = FileSystems.get_filesystem(path)
+ filesystem.report_lineage(path, Lineage.sources(), level=level)
@staticmethod
- def report_sink_lineage(path):
+ def report_sink_lineage(path, level=None):
"""
Report sink :class:`~apache_beam.metrics.metric.Lineage`.
Args:
path: string path to be reported.
+ level: the level of file path. default to
+ :class:`~apache_beam.io.filesystem.FileSystem.Lineage`.FILE.
"""
- FileSystems.get_filesystem(path).report_lineage(path, Lineage.sinks())
+ filesystem = FileSystems.get_filesystem(path)
+ filesystem.report_lineage(path, Lineage.sinks(), level=level)
diff --git a/sdks/python/apache_beam/io/gcp/bigquery.py
b/sdks/python/apache_beam/io/gcp/bigquery.py
index 9f60b5af672..11e0d098b2f 100644
--- a/sdks/python/apache_beam/io/gcp/bigquery.py
+++ b/sdks/python/apache_beam/io/gcp/bigquery.py
@@ -1163,7 +1163,7 @@ class _CustomBigQueryStorageSource(BoundedSource):
self.table_reference.datasetId,
self.table_reference.tableId)
Lineage.sources().add(
- 'bigquery',
+ "bigquery",
self.table_reference.projectId,
self.table_reference.datasetId,
self.table_reference.tableId)
diff --git a/sdks/python/apache_beam/io/gcp/gcsfilesystem.py
b/sdks/python/apache_beam/io/gcp/gcsfilesystem.py
index 3763e21abc9..d48a9ab02b0 100644
--- a/sdks/python/apache_beam/io/gcp/gcsfilesystem.py
+++ b/sdks/python/apache_beam/io/gcp/gcsfilesystem.py
@@ -26,7 +26,6 @@ https://github.com/apache/beam/blob/master/sdks/python/OWNERS
# pytype: skip-file
-import traceback
from typing import BinaryIO # pylint: disable=unused-import
from apache_beam.io.filesystem import BeamIOError
@@ -368,13 +367,14 @@ class GCSFileSystem(FileSystem):
if exceptions:
raise BeamIOError("Delete operation failed", exceptions)
- def report_lineage(self, path, lineage):
+ def report_lineage(self, path, lineage, level=None):
try:
components = gcsio.parse_gcs_path(path, object_optional=True)
except ValueError:
# report lineage is fail-safe
- traceback.print_exc()
return
- if components and not components[-1]:
+ if level == FileSystem.LineageLevel.TOP_LEVEL \
+ or(len(components) > 1 and components[-1] == ''):
+ # bucket only
components = components[:-1]
- lineage.add('gcs', *components, last_segment_sep='/')
+ lineage.add('gcs', *components)
diff --git a/sdks/python/apache_beam/io/gcp/gcsfilesystem_test.py
b/sdks/python/apache_beam/io/gcp/gcsfilesystem_test.py
index ade8529dcac..ec7fa94b05f 100644
--- a/sdks/python/apache_beam/io/gcp/gcsfilesystem_test.py
+++ b/sdks/python/apache_beam/io/gcp/gcsfilesystem_test.py
@@ -382,8 +382,7 @@ class GCSFileSystemTest(unittest.TestCase):
def _verify_lineage(self, uri, expected_segments):
lineage_mock = mock.MagicMock()
self.fs.report_lineage(uri, lineage_mock)
- lineage_mock.add.assert_called_once_with(
- "gcs", *expected_segments, last_segment_sep='/')
+ lineage_mock.add.assert_called_once_with("gcs", *expected_segments)
if __name__ == '__main__':
diff --git a/sdks/python/apache_beam/io/localfilesystem.py
b/sdks/python/apache_beam/io/localfilesystem.py
index daf69b8d030..e9fe7dd4b1c 100644
--- a/sdks/python/apache_beam/io/localfilesystem.py
+++ b/sdks/python/apache_beam/io/localfilesystem.py
@@ -364,6 +364,3 @@ class LocalFileSystem(FileSystem):
if exceptions:
raise BeamIOError("Delete operation failed", exceptions)
-
- def report_lineage(self, path, lineage):
- lineage.add('filesystem', 'localhost', path, last_segment_sep='/')
diff --git a/sdks/python/apache_beam/metrics/cells.py
b/sdks/python/apache_beam/metrics/cells.py
index c2c2e8015ef..a75f2dc3168 100644
--- a/sdks/python/apache_beam/metrics/cells.py
+++ b/sdks/python/apache_beam/metrics/cells.py
@@ -811,9 +811,6 @@ class BoundedTrieData(object):
else:
return False
- def flattened(self):
- return self.as_trie().flattened()
-
def to_proto(self) -> metrics_pb2.BoundedTrie:
return metrics_pb2.BoundedTrie(
bound=self._bound,
@@ -825,9 +822,7 @@ class BoundedTrieData(object):
return BoundedTrieData(
bound=proto.bound,
singleton=tuple(proto.singleton) if proto.singleton else None,
- root=(
- _BoundedTrieNode.from_proto(proto.root)
- if proto.HasField('root') else None))
+ root=_BoundedTrieNode.from_proto(proto.root) if proto.root else None)
def as_trie(self):
if self._root is not None:
diff --git a/sdks/python/apache_beam/metrics/metric.py
b/sdks/python/apache_beam/metrics/metric.py
index 58a74afb9de..33af25e20ca 100644
--- a/sdks/python/apache_beam/metrics/metric.py
+++ b/sdks/python/apache_beam/metrics/metric.py
@@ -33,7 +33,6 @@ from typing import TYPE_CHECKING
from typing import Dict
from typing import FrozenSet
from typing import Iterable
-from typing import Iterator
from typing import List
from typing import Optional
from typing import Set
@@ -339,12 +338,12 @@ class Lineage:
for lineage tracking."""
LINEAGE_NAMESPACE = "lineage"
- SOURCE = "sources_v2"
- SINK = "sinks_v2"
+ SOURCE = "sources"
+ SINK = "sinks"
_METRICS = {
- SOURCE: Metrics.bounded_trie(LINEAGE_NAMESPACE, SOURCE),
- SINK: Metrics.bounded_trie(LINEAGE_NAMESPACE, SINK)
+ SOURCE: Metrics.string_set(LINEAGE_NAMESPACE, SOURCE),
+ SINK: Metrics.string_set(LINEAGE_NAMESPACE, SINK)
}
def __init__(self, label: str) -> None:
@@ -393,32 +392,8 @@ class Lineage:
return ':'.join((system, subtype, segs))
return ':'.join((system, segs))
- @staticmethod
- def _get_fqn_parts(
- system: str,
- *segments: str,
- subtype: Optional[str] = None,
- last_segment_sep: Optional[str] = None) -> Iterator[str]:
- yield system + ':'
- if subtype:
- yield subtype + ':'
- if segments:
- for segment in segments[:-1]:
- yield segment + '.'
- if last_segment_sep:
- sub_segments = segments[-1].split(last_segment_sep)
- for sub_segment in sub_segments[:-1]:
- yield sub_segment + last_segment_sep
- yield sub_segments[-1]
- else:
- yield segments[-1]
-
def add(
- self,
- system: str,
- *segments: str,
- subtype: Optional[str] = None,
- last_segment_sep: Optional[str] = None) -> None:
+ self, system: str, *segments: str, subtype: Optional[str] = None) ->
None:
"""
Adds the given details as Lineage.
@@ -439,35 +414,21 @@ class Lineage:
The first positional argument serves as system, if full segments are
provided, or the full FQN if it is provided as a single argument.
"""
- self.add_raw(
- *self._get_fqn_parts(
- system,
- *segments,
- subtype=subtype,
- last_segment_sep=last_segment_sep))
-
- def add_raw(self, *rollup_segments: str) -> None:
- """Adds the given fqn as lineage.
-
- `rollup_segments` should be an iterable of strings whose concatenation
- is a valid Dataplex FQN. In particular, this means they will often have
- trailing delimiters.
- """
- self.metric.add(rollup_segments)
+ system_or_details = system
+ if len(segments) == 0 and subtype is None:
+ self.metric.add(system_or_details)
+ else:
+ self.metric.add(self.get_fq_name(system, *segments, subtype=subtype))
@staticmethod
- def query(results: MetricResults,
- label: str,
- truncated_marker: str = '*') -> Set[str]:
+ def query(results: MetricResults, label: str) -> Set[str]:
if not label in Lineage._METRICS:
raise ValueError("Label {} does not exist for Lineage", label)
response = results.query(
MetricsFilter().with_namespace(Lineage.LINEAGE_NAMESPACE).with_name(
- label))[MetricResults.BOUNDED_TRIES]
+ label))[MetricResults.STRINGSETS]
result = set()
for metric in response:
- for fqn in metric.committed.flattened():
- result.add(''.join(fqn[:-1]) + (truncated_marker if fqn[-1] else ''))
- for fqn in metric.attempted.flattened():
- result.add(''.join(fqn[:-1]) + (truncated_marker if fqn[-1] else ''))
+ result.update(metric.committed)
+ result.update(metric.attempted)
return result
diff --git a/sdks/python/apache_beam/metrics/metric_test.py
b/sdks/python/apache_beam/metrics/metric_test.py
index 2e2e51b267a..524a2143172 100644
--- a/sdks/python/apache_beam/metrics/metric_test.py
+++ b/sdks/python/apache_beam/metrics/metric_test.py
@@ -271,19 +271,14 @@ class LineageTest(unittest.TestCase):
def test_add(self):
lineage = Lineage(Lineage.SOURCE)
- added = set()
+ stringset = set()
# override
- lineage.metric = added
+ lineage.metric = stringset
lineage.add("s", "1", "2")
lineage.add("s:3.4")
lineage.add("s", "5", "6.7")
lineage.add("s", "1", "2", subtype="t")
- lineage.add("sys", "seg1", "seg2", "seg3/part2/part3",
last_segment_sep='/')
- self.assertSetEqual(
- added,
- {('s:', '1.', '2'), ('s:3.4:', ), ('s:', '5.', '6.7'),
- ('s:', 't:', '1.', '2'),
- ('sys:', 'seg1.', 'seg2.', 'seg3/', 'part2/', 'part3')})
+ self.assertSetEqual(stringset, {"s:1.2", "s:3.4", "s:t:1.2", "s:5.`6.7`"})
if __name__ == '__main__':
