diff --git a/sdks/python/apache_beam/internal/cloudpickle_pickler.py b/sdks/python/apache_beam/internal/cloudpickle_pickler.py index cea4f01f803c..9710878f5b67 100644 --- a/sdks/python/apache_beam/internal/cloudpickle_pickler.py +++ b/sdks/python/apache_beam/internal/cloudpickle_pickler.py @@ -29,6 +29,7 @@ import base64 import bz2 +import dataclasses import io import logging import sys @@ -187,7 +188,10 @@ def _dumps( with _pickle_lock: with io.BytesIO() as file: if enable_stable_code_identifier_pickling: - config = STABLE_CODE_IDENTIFIER_CONFIG + config = dataclasses.replace( + config, + get_code_object_params=STABLE_CODE_IDENTIFIER_CONFIG. + get_code_object_params) pickler = cloudpickle.CloudPickler(file, config=config) try: pickler.dispatch_table[type(flags.FLAGS)] = _pickle_absl_flags diff --git a/sdks/python/apache_beam/pipeline.py b/sdks/python/apache_beam/pipeline.py index fea42bbc0cb3..0e551703266d 100644 --- a/sdks/python/apache_beam/pipeline.py +++ b/sdks/python/apache_beam/pipeline.py @@ -1081,6 +1081,8 @@ def to_runner_api( # because other runners do not have this option. context.enable_best_effort_deterministic_pickling = getattr( self.runner, 'enable_best_effort_deterministic_pickling', False) + context.enable_stable_identifier_pickling = getattr( + self.runner, 'enable_stable_code_identifier_pickling', False) # The RunnerAPI spec requires certain transforms and side-inputs to have KV # inputs (and corresponding outputs).