[ https://issues.apache.org/jira/browse/BEAM-5132?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Beam JIRA Bot updated BEAM-5132: -------------------------------- Labels: stale-P2 (was: ) > Composite windowing fail with exception: AttributeError: 'NoneType' object > has no attribute 'time' > -------------------------------------------------------------------------------------------------- > > Key: BEAM-5132 > URL: https://issues.apache.org/jira/browse/BEAM-5132 > Project: Beam > Issue Type: Bug > Components: sdk-py-core > Affects Versions: 2.5.0 > Environment: Windows machine > Reporter: Bui Nguyen Thang > Priority: P2 > Labels: stale-P2 > > Tried to apply a window function to the existing pipeline that was running > fine. Got the following error: > {code:java} > 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'] > {code} > 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: > {code:python} > 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) > {code} -- This message was sent by Atlassian Jira (v8.3.4#803005)