File tree Expand file tree Collapse file tree
portability/fn_api_runner Expand file tree Collapse file tree Original file line number Diff line number Diff line change @@ -51,8 +51,7 @@ def test_xlang_parquetio_write(self):
5151 address = 'localhost:%s' % port
5252 try :
5353 with TestPipeline () as p :
54- p .get_pipeline_options ().view_as (DebugOptions ).experiments .append (
55- 'jar_packages=' + expansion_jar )
54+ p .get_pipeline_options ().view_as (DebugOptions ).add_experiment ('jar_packages=' + expansion_jar )
5655 p .not_use_test_runner_api = True
5756 _ = p \
5857 | beam .Create ([
Original file line number Diff line number Diff line change @@ -196,14 +196,7 @@ def test_try_split_with_any_exception(self):
196196class UseSdfBoundedSourcesTests (unittest .TestCase ):
197197 def _run_sdf_wrapper_pipeline (self , source , expected_values ):
198198 with beam .Pipeline () as p :
199- experiments = (p ._options .view_as (DebugOptions ).experiments or [])
200-
201- # Setup experiment option to enable using SDFBoundedSourceWrapper
202- if 'beam_fn_api' not in experiments :
203- # Required so mocking below doesn't mock Create used in assert_that.
204- experiments .append ('beam_fn_api' )
205-
206- p ._options .view_as (DebugOptions ).experiments = experiments
199+ p ._options .view_as (DebugOptions ).add_experiment ('beam_fn_api' )
207200
208201 actual = p | beam .io .Read (source )
209202 assert_that (actual , equal_to (expected_values ))
Original file line number Diff line number Diff line change @@ -233,10 +233,7 @@ def __init__(
233233 # set default experiments for portable runners
234234 # (needs to occur prior to pipeline construction)
235235 if runner .is_fnapi_compatible ():
236- experiments = (self ._options .view_as (DebugOptions ).experiments or [])
237- if not 'beam_fn_api' in experiments :
238- experiments .append ('beam_fn_api' )
239- self ._options .view_as (DebugOptions ).experiments = experiments
236+ self ._options .view_as (DebugOptions ).add_experiment ('beam_fn_api' )
240237
241238 self .local_tempdir = tempfile .mkdtemp (prefix = 'beam-pipeline-temp' )
242239
Original file line number Diff line number Diff line change @@ -146,16 +146,12 @@ def __init__(
146146 ])
147147 # TODO: Use enumerated type instead of strings for job types.
148148 if job_type .startswith ('FNAPI_' ):
149- self .debug_options .experiments = self .debug_options .experiments or []
150-
151- debug_options_experiments = self .debug_options .experiments
152149 # Add use_multiple_sdk_containers flag if it's not already present. Do not
153150 # add the flag if 'no_use_multiple_sdk_containers' is present.
154151 # TODO: Cleanup use_multiple_sdk_containers once we deprecate Python SDK
155152 # till version 2.4.
156- if ('use_multiple_sdk_containers' not in debug_options_experiments and
157- 'no_use_multiple_sdk_containers' not in debug_options_experiments ):
158- debug_options_experiments .append ('use_multiple_sdk_containers' )
153+ if ('no_use_multiple_sdk_containers' not in (self .debug_options .experiments or [])):
154+ self .debug_options .add_experiment ('use_multiple_sdk_containers' )
159155 # FlexRS
160156 if self .google_cloud_options .flexrs_goal == 'COST_OPTIMIZED' :
161157 self .proto .flexResourceSchedulingGoal = (
Original file line number Diff line number Diff line change @@ -146,11 +146,7 @@ def run_pipeline(
146146 RuntimeValueProvider .set_runtime_options ({})
147147
148148 # Setup "beam_fn_api" experiment options if lacked.
149- experiments = (
150- options .view_as (pipeline_options .DebugOptions ).experiments or [])
151- if not 'beam_fn_api' in experiments :
152- experiments .append ('beam_fn_api' )
153- options .view_as (pipeline_options .DebugOptions ).experiments = experiments
149+ options .view_as (pipeline_options .DebugOptions ).add_experiment ('beam_fn_api' )
154150
155151 # This is sometimes needed if type checking is disabled
156152 # to enforce that the inputs (and outputs) of GroupByKey operations
You can’t perform that action at this time.
0 commit comments