Hi Kyle I'm on 2.15. Thanks for pointing me to the JIRA, I'll watch it and also try to see what's causing the problem.
Best regards Pawel On Fri, 13 Sep 2019 at 01:43, Kyle Weaver <kcwea...@google.com> wrote: > Hi Pawel, could you tell us which version of the Beam Python SDK you are > using? > > For the record, this looks like a known issue: > https://issues.apache.org/jira/browse/BEAM-6860 > > Kyle Weaver | Software Engineer | github.com/ibzib | kcwea...@google.com > > > On Wed, Sep 11, 2019 at 6:33 AM Paweł Kordek <pawel.kor...@farfetch.com> > wrote: > >> Hi >> >> I was developing a simple pipeline where I aggregate records by key and >> sum values for a predefined window. I was getting some errors, and after >> checking, I am getting exactly the same issues when running Wikipedia >> example from the Beam repo. The output is as follows: >> ------------------------------------------- >> INFO:root:Missing pipeline option (runner). Executing pipeline using the >> default runner: DirectRunner. >> INFO:root:==================== <function annotate_downstream_side_inputs >> at 0x7f333fc1fe60> ==================== >> INFO:root:==================== <function fix_side_input_pcoll_coders at >> 0x7f333fc1ff80> ==================== >> INFO:root:==================== <function lift_combiners at >> 0x7f333fc1d050> ==================== >> INFO:root:==================== <function expand_sdf at 0x7f333fc1d0e0> >> ==================== >> INFO:root:==================== <function expand_gbk at 0x7f333fc1d170> >> ==================== >> INFO:root:==================== <function sink_flattens at 0x7f333fc1d290> >> ==================== >> INFO:root:==================== <function greedily_fuse at 0x7f333fc1d320> >> ==================== >> INFO:root:==================== <function read_to_impulse at >> 0x7f333fc1d3b0> ==================== >> INFO:root:==================== <function impulse_to_input at >> 0x7f333fc1d440> ==================== >> INFO:root:==================== <function inject_timer_pcollections at >> 0x7f333fc1d5f0> ==================== >> INFO:root:==================== <function sort_stages at 0x7f333fc1d680> >> ==================== >> INFO:root:==================== <function window_pcollection_coders at >> 0x7f333fc1d710> ==================== >> INFO:root:Running >> ((((((ref_AppliedPTransform_ReadFromText/Read_3)+(ref_AppliedPTransform_ComputeTopSessions/ExtractUserAndTimestamp_5))+(ref_AppliedPTransform_ComputeTopSessions/Filter(<lambda >> at >> top_wikipedia_sessions.py:127>)_6))+(ref_AppliedPTransform_ComputeTopSessions/ComputeSessions/ComputeSessionsWindow_8))+(ref_AppliedPTransform_ComputeTopSessions/ComputeSessions/PerElement/PerElement:PairWithVoid_10))+(ComputeTopSessions/ComputeSessions/PerElement/CombinePerKey(CountCombineFn)/Precombine))+(ComputeTopSessions/ComputeSessions/PerElement/CombinePerKey(CountCombineFn)/Group/Write) >> INFO:root:Running >> (((((((ComputeTopSessions/ComputeSessions/PerElement/CombinePerKey(CountCombineFn)/Group/Read)+(ComputeTopSessions/ComputeSessions/PerElement/CombinePerKey(CountCombineFn)/Merge))+(ComputeTopSessions/ComputeSessions/PerElement/CombinePerKey(CountCombineFn)/ExtractOutputs))+(ref_AppliedPTransform_ComputeTopSessions/SessionsToStrings_18))+(ref_AppliedPTransform_ComputeTopSessions/TopPerMonth/TopPerMonthWindow_20))+(ref_AppliedPTransform_ComputeTopSessions/TopPerMonth/Top/KeyWithVoid_22))+(ComputeTopSessions/TopPerMonth/Top/CombinePerKey/Precombine))+(ComputeTopSessions/TopPerMonth/Top/CombinePerKey/Group/Write) >> INFO:root:Running >> (((ref_AppliedPTransform_WriteToText/Write/WriteImpl/DoOnce/Read_36)+(ref_AppliedPTransform_WriteToText/Write/WriteImpl/InitializeWrite_37))+(ref_PCollection_PCollection_19/Write))+(ref_PCollection_PCollection_20/Write) >> INFO:root:Running >> ((((((((ComputeTopSessions/TopPerMonth/Top/CombinePerKey/Group/Read)+(ComputeTopSessions/TopPerMonth/Top/CombinePerKey/Merge))+(ComputeTopSessions/TopPerMonth/Top/CombinePerKey/ExtractOutputs))+(ref_AppliedPTransform_ComputeTopSessions/TopPerMonth/Top/UnKey_30))+(ref_AppliedPTransform_ComputeTopSessions/FormatOutput_31))+(ref_AppliedPTransform_WriteToText/Write/WriteImpl/WriteBundles_38))+(ref_AppliedPTransform_WriteToText/Write/WriteImpl/Pair_39))+(ref_AppliedPTransform_WriteToText/Write/WriteImpl/WindowInto(WindowIntoFn)_40))+(WriteToText/Write/WriteImpl/GroupByKey/Write) >> Traceback (most recent call last): >> File "apache_beam/runners/common.py", line 829, in >> apache_beam.runners.common.DoFnRunner._invoke_bundle_method >> File "apache_beam/runners/common.py", line 403, in >> apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle >> File "apache_beam/runners/common.py", line 406, in >> apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle >> File "apache_beam/runners/common.py", line 982, in >> apache_beam.runners.common._OutputProcessor.finish_bundle_outputs >> File "apache_beam/runners/worker/operations.py", line 142, in >> apache_beam.runners.worker.operations.SingletonConsumerSet.receive >> File "apache_beam/runners/worker/operations.py", line 122, in >> apache_beam.runners.worker.operations.ConsumerSet.update_counters_start >> File "apache_beam/runners/worker/opcounters.py", line 196, in >> apache_beam.runners.worker.opcounters.OperationCounters.update_from >> File "apache_beam/runners/worker/opcounters.py", line 214, in >> apache_beam.runners.worker.opcounters.OperationCounters.do_sample >> File "apache_beam/coders/coder_impl.py", line 1014, in >> apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables >> File "apache_beam/coders/coder_impl.py", line 1030, in >> apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables >> File "apache_beam/coders/coder_impl.py", line 814, in >> apache_beam.coders.coder_impl.SequenceCoderImpl.estimate_size >> File "apache_beam/coders/coder_impl.py", line 828, 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 494, in >> apache_beam.coders.coder_impl.IntervalWindowCoderImpl.estimate_size >> TypeError: Cannot convert GlobalWindow to >> apache_beam.utils.windowed_value._IntervalWindowBase >> >> During handling of the above exception, another exception occurred: >> >> Traceback (most recent call last): >> File "top_wikipedia_sessions.py", line 171, in <module> >> run() >> File "top_wikipedia_sessions.py", line 166, in run >> | WriteToText(known_args.output)) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/pipeline.py", >> line 426, in __exit__ >> self.run().wait_until_finish() >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/pipeline.py", >> line 406, in run >> self._options).run(False) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/pipeline.py", >> line 419, in run >> return self.runner.run_pipeline(self, self._options) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/direct/direct_runner.py", >> line 129, in run_pipeline >> return runner.run_pipeline(pipeline, options) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner.py", >> line 366, in run_pipeline >> default_environment=self._default_environment)) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner.py", >> line 373, in run_via_runner_api >> return self.run_stages(stage_context, stages) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner.py", >> line 455, in run_stages >> stage_context.safe_coders) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner.py", >> line 733, in _run_stage >> result, splits = bundle_manager.process_bundle(data_input, >> data_output) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner.py", >> line 1688, in process_bundle >> part, expected_outputs), part_inputs): >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/concurrent/futures/_base.py", >> line 598, in result_iterator >> yield fs.pop().result() >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/concurrent/futures/_base.py", >> line 435, in result >> return self.__get_result() >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/concurrent/futures/_base.py", >> line 384, in __get_result >> raise self._exception >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/concurrent/futures/thread.py", >> line 57, in run >> result = self.fn(*self.args, **self.kwargs) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner.py", >> line 1688, in <lambda> >> part, expected_outputs), part_inputs): >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner.py", >> line 1626, in process_bundle >> result_future = >> self._worker_handler.control_conn.push(process_bundle_req) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner.py", >> line 1080, in push >> response = self.worker.do_instruction(request) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", >> line 343, in do_instruction >> request.instruction_id) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", >> line 369, in process_bundle >> bundle_processor.process_bundle(instruction_id)) >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", >> line 598, in process_bundle >> op.finish() >> File "apache_beam/runners/worker/operations.py", line 611, in >> apache_beam.runners.worker.operations.DoOperation.finish >> File "apache_beam/runners/worker/operations.py", line 612, in >> apache_beam.runners.worker.operations.DoOperation.finish >> File "apache_beam/runners/worker/operations.py", line 613, in >> apache_beam.runners.worker.operations.DoOperation.finish >> File "apache_beam/runners/common.py", line 847, in >> apache_beam.runners.common.DoFnRunner.finish >> File "apache_beam/runners/common.py", line 831, in >> apache_beam.runners.common.DoFnRunner._invoke_bundle_method >> File "apache_beam/runners/common.py", line 872, in >> apache_beam.runners.common.DoFnRunner._reraise_augmented >> File >> "/home/pawel/miniconda3/envs/beam/lib/python3.7/site-packages/future/utils/__init__.py", >> line 421, in raise_with_traceback >> raise exc.with_traceback(traceback) >> File "apache_beam/runners/common.py", line 829, in >> apache_beam.runners.common.DoFnRunner._invoke_bundle_method >> File "apache_beam/runners/common.py", line 403, in >> apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle >> File "apache_beam/runners/common.py", line 406, in >> apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle >> File "apache_beam/runners/common.py", line 982, in >> apache_beam.runners.common._OutputProcessor.finish_bundle_outputs >> File "apache_beam/runners/worker/operations.py", line 142, in >> apache_beam.runners.worker.operations.SingletonConsumerSet.receive >> File "apache_beam/runners/worker/operations.py", line 122, in >> apache_beam.runners.worker.operations.ConsumerSet.update_counters_start >> File "apache_beam/runners/worker/opcounters.py", line 196, in >> apache_beam.runners.worker.opcounters.OperationCounters.update_from >> File "apache_beam/runners/worker/opcounters.py", line 214, in >> apache_beam.runners.worker.opcounters.OperationCounters.do_sample >> File "apache_beam/coders/coder_impl.py", line 1014, in >> apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables >> File "apache_beam/coders/coder_impl.py", line 1030, in >> apache_beam.coders.coder_impl.WindowedValueCoderImpl.get_estimated_size_and_observables >> File "apache_beam/coders/coder_impl.py", line 814, in >> apache_beam.coders.coder_impl.SequenceCoderImpl.estimate_size >> File "apache_beam/coders/coder_impl.py", line 828, 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 494, in >> apache_beam.coders.coder_impl.IntervalWindowCoderImpl.estimate_size >> TypeError: Cannot convert GlobalWindow to >> apache_beam.utils.windowed_value._IntervalWindowBase [while running >> 'WriteToText/Write/WriteImpl/WriteBundles'] >> ------------------------------------------------------- >> >> To run it I've downloaded a single json with Wiki data and run is as >> follows (running from BEAM_REPO/sdks/python/apache_beam/examples/complete >> >> *python top_wikipedia_sessions.py --input >> /data/wiki/wiki_data-000000000492.json --output /tmp/beam/wiki* >> >> This fails somewhere close to the end, in fact I can find some results >> (possibly complete) in >> */tmp/beam/beam-temp-wiki-5971176ad49411e9b16448ba4ef75ccc/c18aa9d5-221f-4dc3-bc1a-83c101ee54ba.wiki* >> >> I tried to find the exact cause in the code but I don't understand Beam's >> codebase enough, would appreciate some hints/explanations. There is a >> question >> on SO >> <https://stackoverflow.com/questions/54745869/how-to-create-a-dataflow-pipeline-from-pub-sub-to-gcs-in-python/54791913#54791913>, >> but that doesn't help much in explaining why this error is observed. >> >> Thanks! >> Pawel >> >> This email and any files transmitted with it are confidential and >> intended solely for the use of the individual or entity to whom they are >> addressed. If you have received this email in error please notify the >> system manager. This message contains confidential information and is >> intended only for the individual named. If you are not the named addressee >> you should not disseminate, distribute or copy this e-mail. Please notify >> the sender immediately by e-mail if you have received this e-mail by >> mistake and delete this e-mail from your system. If you are not the >> intended recipient you are notified that disclosing, copying, distributing >> or taking any action in reliance on the contents of this information is >> strictly prohibited. >> > -- This email and any files transmitted with it are confidential and intended solely for the use of the individual or entity to whom they are addressed. If you have received this email in error please notify the system manager. This message contains confidential information and is intended only for the individual named. If you are not the named addressee you should not disseminate, distribute or copy this e-mail. Please notify the sender immediately by e-mail if you have received this e-mail by mistake and delete this e-mail from your system. If you are not the intended recipient you are notified that disclosing, copying, distributing or taking any action in reliance on the contents of this information is strictly prohibited.
