[
https://issues.apache.org/jira/browse/BEAM-6860?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16929568#comment-16929568
]
Ryan Garver commented on BEAM-6860:
-----------------------------------
I'm seeing this on Beam 2.15.0 and Python 3.7.4. The code is a really simple
test of windowing in the {{DirectRunner}} (read from a file, window, aggregate,
write to a file). I've been peeling back layers of this and the error
definitely happens in {{WriteToText}} and is caused by the presence of a
{{beam.WindowInto}} in the pipeline.
Any thoughts on what is going on and how it might be worked around would be
helpful.
> WriteToText crash with "GlobalWindow -> ._IntervalWindowBase"
> -------------------------------------------------------------
>
> Key: BEAM-6860
> URL: https://issues.apache.org/jira/browse/BEAM-6860
> Project: Beam
> Issue Type: Bug
> Components: sdk-py-core
> Affects Versions: 2.11.0
> Environment: macOS, DirectRunner, python 2.7.15 via
> pyenv/pyenv-virtualenv
> Reporter: Henrik
> Priority: Critical
> Labels: newbie
>
> Main error:
> > Cannot convert GlobalWindow to
> > apache_beam.utils.windowed_value._IntervalWindowBase
> This is very hard for me to debug. Doing a DoPar call before, printing the
> input, gives me just what I want; so the lines of data to serialise are
> "alright"; just JSON strings, in fact.
> Stacktrace:
> {code:java}
> Traceback (most recent call last):
> File "./okr_end_ride.py", line 254, in <module>
> run()
> File "./okr_end_ride.py", line 250, in run
> run_pipeline(pipeline_options, known_args)
> File "./okr_end_ride.py", line 198, in run_pipeline
> | 'write_all' >> WriteToText(known_args.output,
> file_name_suffix=".txt")
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/pipeline.py",
> line 426, in __exit__
> self.run().wait_until_finish()
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/pipeline.py",
> line 406, in run
> self._options).run(False)
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/pipeline.py",
> line 419, in run
> return self.runner.run_pipeline(self, self._options)
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/direct/direct_runner.py",
> line 132, in run_pipeline
> return runner.run_pipeline(pipeline, options)
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/portability/fn_api_runner.py",
> line 275, in run_pipeline
> default_environment=self._default_environment))
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/portability/fn_api_runner.py",
> line 278, in run_via_runner_api
> return self.run_stages(*self.create_stages(pipeline_proto))
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/portability/fn_api_runner.py",
> line 354, in run_stages
> stage_context.safe_coders)
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/portability/fn_api_runner.py",
> line 509, in run_stage
> data_input, data_output)
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/portability/fn_api_runner.py",
> line 1206, in process_bundle
> result_future = self._controller.control_handler.push(process_bundle)
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/portability/fn_api_runner.py",
> line 821, in push
> response = self.worker.do_instruction(request)
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 265, in do_instruction
> request.instruction_id)
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/worker/sdk_worker.py",
> line 281, in process_bundle
> delayed_applications = bundle_processor.process_bundle(instruction_id)
> File
> "/Users/h/.pyenv/versions/2.7.15/envs/log-analytics/lib/python2.7/site-packages/apache_beam/runners/worker/bundle_processor.py",
> line 552, in process_bundle
> op.finish()
> File "apache_beam/runners/worker/operations.py", line 549, in
> apache_beam.runners.worker.operations.DoOperation.finish
> File "apache_beam/runners/worker/operations.py", line 550, in
> apache_beam.runners.worker.operations.DoOperation.finish
> File "apache_beam/runners/worker/operations.py", line 551, in
> apache_beam.runners.worker.operations.DoOperation.finish
> File "apache_beam/runners/common.py", line 758, in
> apache_beam.runners.common.DoFnRunner.finish
> File "apache_beam/runners/common.py", line 752, in
> apache_beam.runners.common.DoFnRunner._invoke_bundle_method
> File "apache_beam/runners/common.py", line 777, in
> apache_beam.runners.common.DoFnRunner._reraise_augmented
> File "apache_beam/runners/common.py", line 750, in
> apache_beam.runners.common.DoFnRunner._invoke_bundle_method
> File "apache_beam/runners/common.py", line 361, in
> apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle
> File "apache_beam/runners/common.py", line 364, in
> apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle
> File "apache_beam/runners/common.py", line 887, in
> apache_beam.runners.common._OutputProcessor.finish_bundle_outputs
> File "apache_beam/runners/worker/operations.py", line 130, in
> apache_beam.runners.worker.operations.SingletonConsumerSet.receive
> File "apache_beam/runners/worker/operations.py", line 110, in
> apache_beam.runners.worker.operations.ConsumerSet.update_counters_start
> File "apache_beam/runners/worker/opcounters.py", line 195, in
> apache_beam.runners.worker.opcounters.OperationCounters.update_from
> File "apache_beam/runners/worker/opcounters.py", line 213, in
> apache_beam.runners.worker.opcounters.OperationCounters.do_sample
> File "apache_beam/coders/coder_impl.py", line 991, in
> apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
> File "apache_beam/coders/coder_impl.py", line 1007, in
> apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables
> File "apache_beam/coders/coder_impl.py", line 791, in
> apache_beam.coders.coder_impl.SequenceCoderImpl.estimate_size
> File "apache_beam/coders/coder_impl.py", line 805, in
> apache_beam.coders.coder_impl.SequenceCoderImpl.get_estimated_size_and_observables
> File "apache_beam/coders/coder_impl.py", line 145, in
> apache_beam.coders.coder_impl.CoderImpl.get_estimated_size_and_observables
> File "apache_beam/coders/coder_impl.py", line 487, in
> apache_beam.coders.coder_impl.IntervalWindowCoderImpl.estimate_size
> TypeError: Cannot convert GlobalWindow to
> apache_beam.utils.windowed_value._IntervalWindowBase [while running
> 'write_all/Write/WriteImpl/WriteBundles']
> {code}
> Slack thread:
> [https://the-asf.slack.com/archives/C9H0YNP3P/p1552928806101200] including
> code snippet
> Your issue tracker seems to be missing components for both the python on go
> SDKs.
>
--
This message was sent by Atlassian Jira
(v8.3.2#803003)
