You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2025/10/03 16:04:36 UTC
(beam) 01/01: Add labels to unlabeled transforms
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch users/damccorm/updateCompat
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 729a8d5d5b6016fb0f7e26d0a745d8cbe5ef2906
Author: Danny Mccormick <da...@google.com>
AuthorDate: Fri Oct 3 12:04:17 2025 -0400
Add labels to unlabeled transforms
---
sdks/python/apache_beam/transforms/core.py | 15 +++++++++------
sdks/python/apache_beam/transforms/external_java.py | 3 ++-
sdks/python/apache_beam/transforms/util.py | 19 +++++++++++--------
3 files changed, 22 insertions(+), 15 deletions(-)
diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py
index cbd78d8222e..4fb764c83ed 100644
--- a/sdks/python/apache_beam/transforms/core.py
+++ b/sdks/python/apache_beam/transforms/core.py
@@ -2352,10 +2352,10 @@ class _ExceptionHandlingWrapper(ptransform.PTransform):
return pcoll
input_count_view = pcoll | 'CountTotal' >> (
- MaybeWindow() | Map(lambda _: 1)
+ MaybeWindow() | "Map(<lambda at core.py:2346>)" >> Map(lambda _: 1)
| CombineGlobally(sum).as_singleton_view())
bad_count_pcoll = result[self._dead_letter_tag] | 'CountBad' >> (
- MaybeWindow() | Map(lambda _: 1)
+ MaybeWindow() | "Map(<lambda at core.py:2349>)" >> Map(lambda _: 1)
| CombineGlobally(sum).without_defaults())
def check_threshold(bad, total, threshold, window=DoFn.WindowParam):
@@ -3521,7 +3521,8 @@ class GroupBy(PTransform):
input_type = pcoll.element_type or typing.Any
return (
pcoll
- | Map(lambda x: (self._key_func()(x), x)).with_output_types(
+ | "Map(<lambda at core.py:3503>)" >>
+ Map(lambda x: (self._key_func()(x), x)).with_output_types(
typehints.Tuple[self._key_type_hint(input_type), input_type])
| GroupByKey())
@@ -3578,7 +3579,8 @@ class _GroupAndAggregate(PTransform):
return (
pcoll
- | Map(lambda x: (key_func(x), value_func(x))).with_output_types(
+ | "Map(<lambda at core.py:3560>)" >>
+ Map(lambda x: (key_func(x), value_func(x))).with_output_types(
typehints.Tuple[key_type_hint, typing.Any])
| CombinePerKey(
TupleCombineFn(
@@ -3599,7 +3601,7 @@ class Select(PTransform):
is the same as
- pcoll | beam.Map(lambda x: beam.Row(a=x.a, b=foo(x)))
+ pcoll | 'label' >> beam.Map(lambda x: beam.Row(a=x.a, b=foo(x)))
"""
def __init__(
self,
@@ -3622,7 +3624,8 @@ class Select(PTransform):
def expand(self, pcoll):
return (
- _MaybePValueWithErrors(pcoll, self._exception_handling_args) | Map(
+ _MaybePValueWithErrors(pcoll, self._exception_handling_args)
+ | "Map(<lambda at core.py:3604>)" >> Map(
lambda x: pvalue.Row(
**{
name: expr(x)
diff --git a/sdks/python/apache_beam/transforms/external_java.py b/sdks/python/apache_beam/transforms/external_java.py
index ebd760f70f7..140ebe8f856 100644
--- a/sdks/python/apache_beam/transforms/external_java.py
+++ b/sdks/python/apache_beam/transforms/external_java.py
@@ -145,7 +145,8 @@ class JavaExternalTransformTest(object):
ImplicitSchemaPayloadBuilder({'data': 'middle'}),
expansion_service)
| beam.ExternalTransform(TEST_COUNT_URN, None, expansion_service)
- | beam.Map(lambda kv: '%s: %s' % kv))
+ | "Map(<lambda at external_java.py:148>)" >> beam.Map(
+ lambda kv: '%s: %s' % kv))
assert_that(res, equal_to(['a: 3', 'b: 1', 'c: 2']))
diff --git a/sdks/python/apache_beam/transforms/util.py b/sdks/python/apache_beam/transforms/util.py
index c63478dc0cf..bbc546b32cd 100644
--- a/sdks/python/apache_beam/transforms/util.py
+++ b/sdks/python/apache_beam/transforms/util.py
@@ -1393,13 +1393,14 @@ def WithKeys(pcoll, k, *args, **kwargs):
if all(isinstance(arg, AsSideInput)
for arg in args) and all(isinstance(kwarg, AsSideInput)
for kwarg in kwargs.values()):
- return pcoll | Map(
+ return pcoll | "Map(<lambda at util.py:1188>)" >> Map(
lambda v, *args, **kwargs: (k(v, *args, **kwargs), v),
*args,
**kwargs)
- return pcoll | Map(lambda v: (k(v, *args, **kwargs), v))
- return pcoll | Map(lambda v: (k(v), v))
- return pcoll | Map(lambda v: (k, v))
+ return pcoll | "Map(<lambda at util.py:1192>)" >> Map(
+ lambda v: (k(v, *args, **kwargs), v))
+ return pcoll | "Map(<lambda at util.py:1193>)" >> Map(lambda v: (k(v), v))
+ return pcoll | "Map(<lambda at util.py:1194>)" >> Map(lambda v: (k, v))
@typehints.with_input_types(tuple[K, V])
@@ -1479,7 +1480,7 @@ class GroupIntoBatches(PTransform):
def expand(self, pcoll):
key_type, value_type = pcoll.element_type.tuple_types
- sharded_pcoll = pcoll | Map(
+ sharded_pcoll = pcoll | "Map(<lambda at util.py:1274>)" >> Map(
lambda key_value: (
ShardedKey(
key_value[0],
@@ -1984,7 +1985,8 @@ class Regex(object):
replacement: the string to be substituted for each match.
"""
regex = Regex._regex_compile(regex)
- return pcoll | Map(lambda elem: regex.sub(replacement, elem))
+ return pcoll | "Map(<lambda at util.py:1779>)" >> Map(
+ lambda elem: regex.sub(replacement, elem))
@staticmethod
@typehints.with_input_types(str)
@@ -2000,7 +2002,8 @@ class Regex(object):
replacement: the string to be substituted for each match.
"""
regex = Regex._regex_compile(regex)
- return pcoll | Map(lambda elem: regex.sub(replacement, elem, 1))
+ return pcoll | "Map(<lambda at util.py:1795>)" >> Map(
+ lambda elem: regex.sub(replacement, elem, 1))
@staticmethod
@typehints.with_input_types(str)
@@ -2091,4 +2094,4 @@ class WaitOn(PTransform):
| f"WaitOn{ix}" >> (beam.FlatMap(lambda x: ()) | GroupByKey()))
for (ix, side) in enumerate(self._to_be_waited_on)
]
- return pcoll | beam.Map(lambda x, *unused_sides: x, *sides)
+ return pcoll | "Map(<lambda at util.py:1886>)" >> beam.Map(lambda x, *unused_sides: x, *sides)