@@ -72,32 +72,36 @@ def __init__(self):
7272 metrics_namespace = self .metrics_namespace ,
7373 is_streaming = False ,
7474 pcollection = 'RunInference/BeamML_RunInference_Postprocess-0.out0' )
75- self .is_streaming = ((self .pipeline .get_option ('mode' ) or
76- 'batch' ) == 'streaming' )
75+ self .opts = self .pipeline .get_pipeline_options ().view_as (
76+ TableRowInferenceOptions )
77+ mode = self .opts .mode or 'batch'
78+ self .is_streaming = mode == 'streaming'
7779 if self .is_streaming :
78- self .subscription = self .pipeline . get_option ( ' input_subscription' )
80+ self .subscription = self .opts . input_subscription
7981
8082 def test (self ):
8183 """Execute the table row inference pipeline for benchmarking."""
82- extra_opts = {}
83-
84- mode = self .pipeline .get_option ('mode' ) or 'batch'
85- extra_opts ['mode' ] = mode
84+ mode = self .opts .mode or 'batch'
85+ extra_opts = {'mode' : mode }
8686
8787 if mode == 'streaming' :
88- extra_opts ['input_subscription' ] = self .pipeline .get_option (
89- 'input_subscription' )
90- extra_opts ['window_size_sec' ] = int (
91- self .pipeline .get_option ('window_size_sec' ) or 60 )
92- extra_opts ['trigger_interval_sec' ] = int (
93- self .pipeline .get_option ('trigger_interval_sec' ) or 30 )
94- else :
95- extra_opts ['input_file' ] = self .pipeline .get_option ('input_file' )
96-
97- for opt in ['output_table' , 'model_path' , 'feature_columns' ]:
98- val = self .pipeline .get_option (opt )
99- if val :
100- extra_opts [opt ] = val
88+ if self .opts .input_subscription :
89+ extra_opts ['input_subscription' ] = self .opts .input_subscription
90+ extra_opts ['window_size_sec' ] = (
91+ self .opts .window_size_sec
92+ if self .opts .window_size_sec is not None else 60 )
93+ extra_opts ['trigger_interval_sec' ] = (
94+ self .opts .trigger_interval_sec
95+ if self .opts .trigger_interval_sec is not None else 30 )
96+ elif self .opts .input_file :
97+ extra_opts ['input_file' ] = self .opts .input_file
98+
99+ if self .opts .output_table :
100+ extra_opts ['output_table' ] = self .opts .output_table
101+ if self .opts .model_path :
102+ extra_opts ['model_path' ] = self .opts .model_path
103+ if self .opts .feature_columns :
104+ extra_opts ['feature_columns' ] = self .opts .feature_columns
101105
102106 self .result = table_row_inference .run (
103107 self .pipeline .get_full_options_as_args (** extra_opts ),
0 commit comments