[
https://issues.apache.org/jira/browse/BEAM-5132?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16583311#comment-16583311
]
Subhash edited comment on BEAM-5132 at 8/17/18 3:34 AM:
--------------------------------------------------------
Hi [~altay], [~thangbui]:
You'll get this issue if you try running the simple example below (adapted from
Apache Beam's windowed_wordcount python example). Hope this helps! :)
{code:java}
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""A streaming word-counting workflow.
Important: streaming pipeline support in Python Dataflow is in development
and is not yet available for use.
"""
from __future__ import absolute_import
import logging
import time
from datetime import datetime
from apache_beam.transforms.trigger import Repeatedly, AfterAny, AfterCount,
AfterProcessingTime, AccumulationMode
from past.builtins import unicode
import apache_beam as beam
import apache_beam.transforms.window as window
def find_words(element):
print element
import re
return re.findall(r'[A-Za-z\']+', element)
class FormatDoFn(beam.DoFn):
def process(self, element, window=beam.DoFn.WindowParam):
ts_format = '%Y-%m-%d %H:%M:%S.%f UTC'
print window.start
print window.end
window_start = window.start.to_utc_datetime().strftime(ts_format)
window_end = window.end.to_utc_datetime().strftime(ts_format)
print element[1], window_start, window_end
return [{'word': element[0],
'count': element[1],
'window_start':window_start,
'window_end':window_end}]
def run(argv=None):
"""Build and run the pipeline."""
with beam.Pipeline() as p:
# Read the text from PubSub messages.
lines = p | beam.Create(["a" + str(i) for i in range(1, 1000)])
# Get the number of appearances of a word.
def count_ones(word_ones):
(word, ones) = word_ones
return (word, sum(ones))
class AddTimestampDoFn(beam.DoFn):
def process(self, element):
print 'Adding timestamp.'
# Extract the numeric Unix seconds-since-epoch timestamp to be
# associated with the current log entry.
timestamp = int(time.mktime(datetime.utcnow().timetuple()))
# Wrap and emit the current entry and new timestamp in a
# TimestampedValue.
yield beam.window.TimestampedValue(element, timestamp)
transformed = (lines
| 'Timestamp' >> beam.ParDo(AddTimestampDoFn())
| 'Batch input' >>
beam.WindowInto(window.FixedWindows(1 * 60),
trigger= Repeatedly(
AfterAny(
AfterCount(3),
AfterProcessingTime(delay=1 * 60))),
accumulation_mode=AccumulationMode.DISCARDING)
| 'Split' >> (beam.FlatMap(find_words)
.with_output_types(unicode))
| 'PairWithOne' >> beam.Map(lambda x: (x, 1))
| 'Group' >> beam.GroupByKey()
| 'Count' >> beam.Map(count_ones)
| 'Format' >> beam.ParDo(FormatDoFn()))
print transformed
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
run()
{code}
was (Author: subby):
[~altay], [~thangbui]:
You'll get this issue if you try running this simple example (adapted from
Apache Beam's windowed_wordcount python example):
{code:java}
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""A streaming word-counting workflow.
Important: streaming pipeline support in Python Dataflow is in development
and is not yet available for use.
"""
from __future__ import absolute_import
import logging
import time
from datetime import datetime
from apache_beam.transforms.trigger import Repeatedly, AfterAny, AfterCount,
AfterProcessingTime, AccumulationMode
from past.builtins import unicode
import apache_beam as beam
import apache_beam.transforms.window as window
def find_words(element):
print element
import re
return re.findall(r'[A-Za-z\']+', element)
class FormatDoFn(beam.DoFn):
def process(self, element, window=beam.DoFn.WindowParam):
ts_format = '%Y-%m-%d %H:%M:%S.%f UTC'
print window.start
print window.end
window_start = window.start.to_utc_datetime().strftime(ts_format)
window_end = window.end.to_utc_datetime().strftime(ts_format)
print element[1], window_start, window_end
return [{'word': element[0],
'count': element[1],
'window_start':window_start,
'window_end':window_end}]
def run(argv=None):
"""Build and run the pipeline."""
with beam.Pipeline() as p:
# Read the text from PubSub messages.
lines = p | beam.Create(["a" + str(i) for i in range(1, 1000)])
# Get the number of appearances of a word.
def count_ones(word_ones):
(word, ones) = word_ones
return (word, sum(ones))
class AddTimestampDoFn(beam.DoFn):
def process(self, element):
print 'Adding timestamp.'
# Extract the numeric Unix seconds-since-epoch timestamp to be
# associated with the current log entry.
timestamp = int(time.mktime(datetime.utcnow().timetuple()))
# Wrap and emit the current entry and new timestamp in a
# TimestampedValue.
yield beam.window.TimestampedValue(element, timestamp)
transformed = (lines
| 'Timestamp' >> beam.ParDo(AddTimestampDoFn())
| 'Batch input' >>
beam.WindowInto(window.FixedWindows(1 * 60),
trigger= Repeatedly(
AfterAny(
AfterCount(3),
AfterProcessingTime(delay=1 * 60))),
accumulation_mode=AccumulationMode.DISCARDING)
| 'Split' >> (beam.FlatMap(find_words)
.with_output_types(unicode))
| 'PairWithOne' >> beam.Map(lambda x: (x, 1))
| 'Group' >> beam.GroupByKey()
| 'Count' >> beam.Map(count_ones)
| 'Format' >> beam.ParDo(FormatDoFn()))
print transformed
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
run()
{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
> 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)
