Details
-
Bug
-
Status: Open
-
P3
-
Resolution: Unresolved
-
2.5.0
-
None
-
None
-
Windows machine
Description
Tried to apply a window function to the existing pipeline that was running fine. Got the following error:
Traceback (most recent call last): File "E:\soft\ide\PyCharm 2018.1.2\helpers\pydev\pydevd.py", line 1664, in <module> main() File "E:\soft\ide\PyCharm 2018.1.2\helpers\pydev\pydevd.py", line 1658, in main globals = debugger.run(setup['file'], None, None, is_module) File "E:\soft\ide\PyCharm 2018.1.2\helpers\pydev\pydevd.py", line 1068, in run pydev_imports.execfile(file, globals, locals) # execute the script File "E:/work/source/ai-data-pipeline-research/metric_pipeline/batch_beam/batch_pipeline_main.py", line 97, in <module> result.wait_until_finish() File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\direct\direct_runner.py", line 421, in wait_until_finish self._executor.await_completion() File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\direct\executor.py", line 398, in await_completion self._executor.await_completion() File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\direct\executor.py", line 444, in await_completion six.reraise(t, v, tb) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\direct\executor.py", line 341, in call finish_state) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\direct\executor.py", line 378, in attempt_call evaluator.process_element(value) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\direct\transform_evaluator.py", line 574, in process_element self.runner.process(element) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\common.py", line 577, in process self._reraise_augmented(exn) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\common.py", line 618, in _reraise_augmented six.reraise(type(new_exn), new_exn, original_traceback) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\common.py", line 575, in process self.do_fn_invoker.invoke_process(windowed_value) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\common.py", line 353, in invoke_process windowed_value, self.process_method(windowed_value.value)) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\runners\common.py", line 651, in process_outputs for result in results: File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\transforms\trigger.py", line 942, in process_entire_key state, windowed_values, output_watermark): File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\transforms\trigger.py", line 1098, in process_elements self.trigger_fn.on_element(value, window, context) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\transforms\trigger.py", line 488, in on_element self.underlying.on_element(element, window, context) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\transforms\trigger.py", line 535, in on_element trigger.on_element(element, window, self._sub_context(context, ix)) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\transforms\trigger.py", line 286, in on_element '', TimeDomain.REAL_TIME, context.get_current_time() + self.delay) File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\transforms\trigger.py", line 728, in get_current_time return self._outer.get_current_time() File "C:\Users\thang\.virtualenvs\metric_pipeline-Mpf2nmv6\lib\site-packages\apache_beam\transforms\trigger.py", line 702, in get_current_time return self._clock.time() AttributeError: 'NoneType' object has no attribute 'time' [while running 'combine all sharpe_ratio to list/CombinePerKey/GroupByKey/GroupByWindow']
the composite window function is copied from official documents:
https://beam.apache.org/documentation/programming-guide/#composite-triggers
Please refer to pipeline relevant source code below:
windowing = beam.WindowInto(FixedWindows(1 * 60), trigger=Repeatedly(AfterAny(AfterCount(100), AfterProcessingTime(delay=1 * 60))), accumulation_mode=AccumulationMode.DISCARDING) valuesPCollection \ | 'calculate sharpe_ratio' >> beam.FlatMap(fn_calculate_sharpe_ratio) \ | 'window sharpe_ratio' >> windowing \ | 'combine all sharpe_ratio to list' >> beam.CombineGlobally(CombineAllToListFn()).without_defaults() \ | 'store sharpe_ratio' >> beam.FlatMap(store_metric_with_now_ts)