From e0f68130f0ec2ac4173a7ee2a38bbb9f13e6c666 Mon Sep 17 00:00:00 2001 From: Ivy Xu Date: Wed, 1 Jul 2026 12:06:49 +0800 Subject: [PATCH 1/2] Remove the workaround for pickle --- sdks/python/apache_beam/examples/complete/game/game_stats.py | 4 +--- .../apache_beam/examples/complete/game/hourly_team_score.py | 4 +--- .../python/apache_beam/examples/complete/game/leader_board.py | 4 +--- sdks/python/apache_beam/examples/complete/game/user_score.py | 4 +--- sdks/python/apache_beam/examples/wordcount_debugging.py | 4 +--- sdks/python/apache_beam/examples/wordcount_with_metrics.py | 4 +--- 6 files changed, 6 insertions(+), 18 deletions(-) diff --git a/sdks/python/apache_beam/examples/complete/game/game_stats.py b/sdks/python/apache_beam/examples/complete/game/game_stats.py index 233d22b75427..5623a66c8211 100644 --- a/sdks/python/apache_beam/examples/complete/game/game_stats.py +++ b/sdks/python/apache_beam/examples/complete/game/game_stats.py @@ -105,9 +105,7 @@ class ParseGameEventFn(beam.DoFn): The human-readable time string is not used here. """ def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.num_parse_errors = Metrics.counter(self.__class__, 'num_parse_errors') def process(self, elem): diff --git a/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py b/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py index 48a105af527d..2a83d65857d2 100644 --- a/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py +++ b/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py @@ -105,9 +105,7 @@ class ParseGameEventFn(beam.DoFn): The human-readable time string is not used here. """ def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.num_parse_errors = Metrics.counter(self.__class__, 'num_parse_errors') def process(self, elem): diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board.py b/sdks/python/apache_beam/examples/complete/game/leader_board.py index 308e1e1cf5c0..a5e6a29e2720 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board.py @@ -114,9 +114,7 @@ class ParseGameEventFn(beam.DoFn): The human-readable time string is not used here. """ def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.num_parse_errors = Metrics.counter(self.__class__, 'num_parse_errors') def process(self, elem): diff --git a/sdks/python/apache_beam/examples/complete/game/user_score.py b/sdks/python/apache_beam/examples/complete/game/user_score.py index 03f0d00fc30f..e878b3f26227 100644 --- a/sdks/python/apache_beam/examples/complete/game/user_score.py +++ b/sdks/python/apache_beam/examples/complete/game/user_score.py @@ -97,9 +97,7 @@ class ParseGameEventFn(beam.DoFn): The human-readable time string is not used here. """ def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.num_parse_errors = Metrics.counter(self.__class__, 'num_parse_errors') def process(self, elem): diff --git a/sdks/python/apache_beam/examples/wordcount_debugging.py b/sdks/python/apache_beam/examples/wordcount_debugging.py index 581bbd3adc1b..9fb446a90a1b 100644 --- a/sdks/python/apache_beam/examples/wordcount_debugging.py +++ b/sdks/python/apache_beam/examples/wordcount_debugging.py @@ -79,9 +79,7 @@ class FilterTextFn(beam.DoFn): """A DoFn that filters for a specific key based on a regular expression.""" def __init__(self, pattern): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.pattern = pattern # A custom metric can track values in your pipeline as it runs. Those # values will be available in the monitoring system of the runner used diff --git a/sdks/python/apache_beam/examples/wordcount_with_metrics.py b/sdks/python/apache_beam/examples/wordcount_with_metrics.py index f575a8d7fbba..e26a627488eb 100644 --- a/sdks/python/apache_beam/examples/wordcount_with_metrics.py +++ b/sdks/python/apache_beam/examples/wordcount_with_metrics.py @@ -53,9 +53,7 @@ class WordExtractingDoFn(beam.DoFn): """Parse each line of input text into words.""" def __init__(self): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.DoFn.__init__(self) + super().__init__() self.words_counter = Metrics.counter(self.__class__, 'words') self.word_lengths_counter = Metrics.counter(self.__class__, 'word_lengths') self.word_lengths_dist = Metrics.distribution( From bb84e402d9dab03c30be53bf2f938b2c1dd2fec7 Mon Sep 17 00:00:00 2001 From: Ivy Xu Date: Wed, 1 Jul 2026 12:22:23 +0800 Subject: [PATCH 2/2] Remove the workaround for pickle --- .../examples/complete/autocomplete.py | 5 ++--- .../examples/complete/game/game_stats.py | 10 ++++------ .../complete/game/hourly_team_score.py | 15 ++++++-------- .../examples/complete/game/leader_board.py | 20 ++++++++----------- .../examples/complete/game/user_score.py | 5 ++--- .../complete/top_wikipedia_sessions.py | 5 ++--- .../examples/cookbook/bigtableio_it_test.py | 4 +--- 7 files changed, 25 insertions(+), 39 deletions(-) diff --git a/sdks/python/apache_beam/examples/complete/autocomplete.py b/sdks/python/apache_beam/examples/complete/autocomplete.py index 4e4c5143b96b..5db2617d6794 100644 --- a/sdks/python/apache_beam/examples/complete/autocomplete.py +++ b/sdks/python/apache_beam/examples/complete/autocomplete.py @@ -58,9 +58,8 @@ def format_result(prefix_candidates): class TopPerPrefix(beam.PTransform): def __init__(self, count): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self._count = count def expand(self, words): diff --git a/sdks/python/apache_beam/examples/complete/game/game_stats.py b/sdks/python/apache_beam/examples/complete/game/game_stats.py index 5623a66c8211..ad63fba1be01 100644 --- a/sdks/python/apache_beam/examples/complete/game/game_stats.py +++ b/sdks/python/apache_beam/examples/complete/game/game_stats.py @@ -129,9 +129,8 @@ class ExtractAndSumScore(beam.PTransform): extracted. """ def __init__(self, field): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.field = field def expand(self, pcoll): @@ -169,9 +168,8 @@ def __init__(self, table_name, dataset, schema, project): schema: Dictionary in the format {'column_name': 'bigquery_type'} project: Name of the Cloud project containing BigQuery table. """ - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.table_name = table_name self.dataset = dataset self.schema = schema diff --git a/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py b/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py index 2a83d65857d2..8542350f0cd5 100644 --- a/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py +++ b/sdks/python/apache_beam/examples/complete/game/hourly_team_score.py @@ -129,9 +129,8 @@ class ExtractAndSumScore(beam.PTransform): extracted. """ def __init__(self, field): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.field = field def expand(self, pcoll): @@ -169,9 +168,8 @@ def __init__(self, table_name, dataset, schema, project): schema: Dictionary in the format {'column_name': 'bigquery_type'} project: Name of the Cloud project containing BigQuery table. """ - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.table_name = table_name self.dataset = dataset self.schema = schema @@ -194,9 +192,8 @@ def expand(self, pcoll): # [START main] class HourlyTeamScore(beam.PTransform): def __init__(self, start_min, stop_min, window_duration): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.start_timestamp = str2timestamp(start_min) self.stop_timestamp = str2timestamp(stop_min) self.window_duration_in_seconds = window_duration * 60 diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board.py b/sdks/python/apache_beam/examples/complete/game/leader_board.py index a5e6a29e2720..96cde175c409 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board.py @@ -138,9 +138,8 @@ class ExtractAndSumScore(beam.PTransform): extracted. """ def __init__(self, field): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.field = field def expand(self, pcoll): @@ -178,9 +177,8 @@ def __init__(self, table_name, dataset, schema, project): schema: Dictionary in the format {'column_name': 'bigquery_type'} project: Name of the Cloud project containing BigQuery table. """ - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.table_name = table_name self.dataset = dataset self.schema = schema @@ -208,9 +206,8 @@ class CalculateTeamScores(beam.PTransform): default. """ def __init__(self, team_window_duration, allowed_lateness): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.team_window_duration = team_window_duration * 60 self.allowed_lateness_seconds = allowed_lateness * 60 @@ -240,9 +237,8 @@ class CalculateUserScores(beam.PTransform): global windowing. Get periodic updates on all users' running scores. """ def __init__(self, allowed_lateness): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.allowed_lateness_seconds = allowed_lateness * 60 def expand(self, pcoll): diff --git a/sdks/python/apache_beam/examples/complete/game/user_score.py b/sdks/python/apache_beam/examples/complete/game/user_score.py index e878b3f26227..ae021b45f560 100644 --- a/sdks/python/apache_beam/examples/complete/game/user_score.py +++ b/sdks/python/apache_beam/examples/complete/game/user_score.py @@ -122,9 +122,8 @@ class ExtractAndSumScore(beam.PTransform): extracted. """ def __init__(self, field): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.field = field def expand(self, pcoll): diff --git a/sdks/python/apache_beam/examples/complete/top_wikipedia_sessions.py b/sdks/python/apache_beam/examples/complete/top_wikipedia_sessions.py index 50b026edf240..440a13fa9a0e 100644 --- a/sdks/python/apache_beam/examples/complete/top_wikipedia_sessions.py +++ b/sdks/python/apache_beam/examples/complete/top_wikipedia_sessions.py @@ -111,9 +111,8 @@ def format_output(element, window=beam.DoFn.WindowParam): class ComputeTopSessions(beam.PTransform): """Computes the top user sessions for each month.""" def __init__(self, sampling_threshold): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() + self.sampling_threshold = sampling_threshold def expand(self, pcoll): diff --git a/sdks/python/apache_beam/examples/cookbook/bigtableio_it_test.py b/sdks/python/apache_beam/examples/cookbook/bigtableio_it_test.py index cc11ec071cc1..84087dee5607 100644 --- a/sdks/python/apache_beam/examples/cookbook/bigtableio_it_test.py +++ b/sdks/python/apache_beam/examples/cookbook/bigtableio_it_test.py @@ -66,9 +66,7 @@ class GenerateTestRows(beam.PTransform): """ def __init__(self, number, project_id=None, instance_id=None, table_id=None): - # TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3. - # super().__init__() - beam.PTransform.__init__(self) + super().__init__() self.number = number self.rand = random.choice(string.ascii_letters + string.digits) self.column_family_id = 'cf1'