apache / apache/beam

Composite windowing fail with exception: AttributeError: 'NoneType' object has no attribute 'time'

Open
#19,070 8 comments 0 reactions 0 assignees View on GitHub
bug core P3 python
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

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
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
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)

```

Imported from Jira [BEAM-5132](https://issues.apache.org/jira/browse/BEAM-5132). Original Jira may contain additional context.
Reported by: thangbui.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.