beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Bui Nguyen Thang (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (BEAM-5132) Composite windowing fail with exception: AttributeError: 'NoneType' object has no attribute 'time'
Date Sun, 12 Aug 2018 22:16:00 GMT

    [ https://issues.apache.org/jira/browse/BEAM-5132?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16577730#comment-16577730
] 

Bui Nguyen Thang commented on BEAM-5132:
----------------------------------------

hi Ahmed,

thanks for prompt feedback
here is how is inited the pipeline
{code:python}
parser = argparse.ArgumentParser()
parser.add_argument('--BIGQUERY_PROJECT_ID')
parser.add_argument('--BIGQUERY_DATASET_ID')

known_args, pipeline_args = parser.parse_known_args()
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = True

if pipeline_options.view_as(GoogleCloudOptions).project is None:
    parser.print_usage()
    print(sys.argv[0] + ': error: argument --project is required')
    sys.exit(1)

pipeline = beam.Pipeline(options=pipeline_options)
{code}

and how i run it in local machine:
{code:python}
--project equity-ai
--BIGQUERY_PROJECT_ID=equity-ai
--BIGQUERY_DATASET_ID=stage_v1
{code}


> 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
>            Assignee: Ahmet Altay
>            Priority: Major
>
> 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
(v7.6.3#76005)

Mime
View raw message