We are currently verifying the patch. Will report back tomorrow. On Thu, Nov 21, 2019 at 8:40 AM Valentyn Tymofieiev <valen...@google.com> wrote:
> That would be helpful, thanks a lot! It should be a straightforward patch. > Also, thanks Guenther, for sharing your investigation on > https://bugs.python.org/issue34572, it was very helpful. > > On Thu, Nov 21, 2019 at 8:25 AM Thomas Weise <t...@apache.org> wrote: > >> Valentyn, thanks a lot for following up on this. >> >> If the change can be cherry picked in isolation, we should be able to >> verify this soon (with 2.16). >> >> >> On Thu, Nov 21, 2019 at 8:12 AM Valentyn Tymofieiev <valen...@google.com> >> wrote: >> >>> To close the loop here: To my knowledge this issue affects all Python 3 >>> users of Portable Flink/Spark runners, and Dataflow Python Streaming users, >>> including users on Python 3.7.3 and newer versions. >>> >>> The issue is addressed on Beam master, and we have a cherry-pick out for >>> Beam 2.17.0. >>> >>> Workaround options for users on 2.16.0 and earlier SDKs: >>> >>> - Patch the SDK you are using with >>> https://github.com/apache/beam/pull/10167. >>> - Temporarily switch to Python 2 until 2.17.0. We have not seen the >>> issue on Python 2, so it may be rare on non-existent on Python 2. >>> - Pass --experiments worker_threads=1 . This option may work only for >>> some, but not all pipelines. >>> >>> See BEAM-8651 <https://issues.apache.org/jira/browse/BEAM-8651> for >>> details on the issue. >>> >>> On Wed, Nov 13, 2019 at 11:55 AM Valentyn Tymofieiev < >>> valen...@google.com> wrote: >>> >>>> I also opened https://issues.apache.org/jira/browse/BEAM-8651 to track >>>> this issue and any recommendation for the users that will come out of it. >>>> >>>> On Thu, Nov 7, 2019 at 6:25 PM Valentyn Tymofieiev <valen...@google.com> >>>> wrote: >>>> >>>>> I think we have heard of this issue from the same source: >>>>> >>>>> This looks exactly like a race condition that we've encountered on >>>>>> Python 3.7.1: There's a bug in some older 3.7.x releases that breaks the >>>>>> thread-safety of the unpickler, as concurrent unpickle threads can >>>>>> access a >>>>>> module before it has been fully imported. See >>>>>> https://bugs.python.org/issue34572 for more information. >>>>>> >>>>>> The traceback shows a Python 3.6 venv so this could be a different >>>>>> issue (the unpickle bug was introduced in version 3.7). If it's the same >>>>>> bug then upgrading to Python 3.7.3 or higher should fix that issue. One >>>>>> potential workaround is to ensure that all of the modules get imported >>>>>> during the initialization of the sdk_worker, as this bug only affects >>>>>> imports done by the unpickler. >>>>> >>>>> >>>>> The symptoms do sound similar, so I would try to reproduce your issue >>>>> on 3.7.3 and see if it is gone, or try to reproduce >>>>> https://bugs.python.org/issue34572 in the version of interpreter you >>>>> use. If this doesn't help, you can try to reproduce the race using your >>>>> input. >>>>> >>>>> To get the output of serialized do fn, you could do the following: >>>>> 1. Patch https://github.com/apache/beam/pull/10036. >>>>> 2. Set logging level to DEBUG, see: >>>>> https://github.com/apache/beam/blob/90d587843172143c15ed392513e396b74569a98c/sdks/python/apache_beam/examples/wordcount.py#L137 >>>>> . >>>>> 3. Check for log output for payload of your transform, it may look >>>>> like: >>>>> >>>>> transforms { >>>>> key: "ref_AppliedPTransform_write/Write/WriteImpl/PreFinalize_42" >>>>> value { >>>>> spec { >>>>> urn: "beam:transform:pardo:v1" >>>>> payload: "\n\347\006\n\275\006\n >>>>> beam:dofn:pickled_python_info:v1\032\230\006eNptU1tPFTEQPgqIFFT >>>>> .... >>>>> >>>>> Then you can extract the output of pickled fn: >>>>> >>>>> from apache_beam.utils import proto_utils >>>>> from apache_beam.portability.api import beam_runner_api_pb2 >>>>> from apache_beam.internal import pickler >>>>> >>>>> payload = b'\n\347\006\n\275\006\n >>>>> beam:dofn:pickled_python_info:v1\032\230\006eNptU1tPFTEQPgqIFFT...' >>>>> pardo_payload = proto_utils.parse_Bytes(x, >>>>> beam_runner_api_pb2.ParDoPayload) >>>>> pickled_fn = pardo_payload.do_fn.spec.payload >>>>> >>>>> pickler.loads(pickle_fn) # Presumably the race happens here when >>>>> unpickling one of your transforms >>>>> (pricingrealtime.aggregation.aggregation_transform). >>>>> >>>>> >>>>> On Wed, Nov 6, 2019 at 10:54 PM Rakesh Kumar <rakeshku...@lyft.com> >>>>> wrote: >>>>> >>>>>> Thanks Valentyn, >>>>>> >>>>>> Aggregation_transform.py doesn't have any transformation method which >>>>>> extends beam.DoFn. We are using plain python method which we passed in >>>>>> beam.Map(). I am not sure how to get the dump of serialized_fn. Can you >>>>>> please let me the process? >>>>>> >>>>>> I also heard that some people ran into this issue on Python 3.7.1 but >>>>>> the same issue is not present on Python 3.7.3. Can you confirm this? >>>>>> >>>>>> >>>>>> >>>>>> On Mon, Oct 28, 2019 at 5:00 PM Valentyn Tymofieiev < >>>>>> valen...@google.com> wrote: >>>>>> >>>>>>> +user@, bcc: dev@ >>>>>>> https://issues.apache.org/jira/browse/BEAM-6158 may be contributing >>>>>>> to this issue, although we saw instances of this bug in exactly opposite >>>>>>> scenarios - when pipeline was defined *in one file*, but not in >>>>>>> multiple files. >>>>>>> >>>>>>> Could you try replacing instances of super() in >>>>>>> aggregation_transform.py as done in >>>>>>> https://github.com/apache/beam/pull/9513 and see if this issue is >>>>>>> still reproducible? >>>>>>> >>>>>>> If that doesn't work, I would try to get the dump of serialized_fn, >>>>>>> and try to reproduce the issue in isolated environment, such as: >>>>>>> >>>>>>> form apache_beam.internal import pickler >>>>>>> serialized_fn = "..content.." >>>>>>> pickler.loads(serialized_fn) >>>>>>> >>>>>>> then I would try to trim the doFn in the example to a >>>>>>> minimally-reproducible example. It could be another issue with dill >>>>>>> dependency. >>>>>>> >>>>>>> >>>>>>> On Mon, Oct 28, 2019 at 2:48 PM Rakesh Kumar <rakeshku...@lyft.com> >>>>>>> wrote: >>>>>>> >>>>>>>> Hi All, >>>>>>>> >>>>>>>> We have noticed a weird intermittent issue on Python3 but we don't >>>>>>>> run into this issue on python2. Sometimes when we are trying to submit >>>>>>>> the >>>>>>>> pipeline, we get AttributeError (Check the stack trace below). we have >>>>>>>> double-checked and we do find the attribute/methods are present in the >>>>>>>> right module and in right place but somehow the pipeline still >>>>>>>> complains >>>>>>>> about it. In some cases, we refer methods before their definition. We >>>>>>>> tried >>>>>>>> to reorder the method definition but that didn't help at all. >>>>>>>> >>>>>>>> We don't see the same issue when the entire pipeline is defined in >>>>>>>> one file. Also, note that this doesn't happen all the time when we >>>>>>>> submit >>>>>>>> the pipeline, so I feel it is some kind of race condition. When we >>>>>>>> enable >>>>>>>> the worker recycle logic it happens most of the time when sdk worker is >>>>>>>> recycled. >>>>>>>> >>>>>>>> Some more information about the environment: >>>>>>>> Python version: 3 >>>>>>>> Beam version: 2.16 >>>>>>>> Flink version: 1.8 >>>>>>>> >>>>>>>> *Stack trace: * >>>>>>>> >>>>>>>> - : >>>>>>>> >>>>>>>> TimerException{java.lang.RuntimeException: Failed to finish remote >>>>>>>> bundle} >>>>>>>> at >>>>>>>> org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$RepeatedTriggerTask.run(SystemProcessingTimeService.java:335) >>>>>>>> at >>>>>>>> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) >>>>>>>> at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) >>>>>>>> at >>>>>>>> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) >>>>>>>> at >>>>>>>> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) >>>>>>>> at >>>>>>>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) >>>>>>>> at >>>>>>>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) >>>>>>>> at java.lang.Thread.run(Thread.java:748) >>>>>>>> Caused by: java.lang.RuntimeException: Failed to finish remote >>>>>>>> bundle >>>>>>>> at >>>>>>>> org.apache.beam.runners.flink.translation.wrappers.streaming.ExecutableStageDoFnOperator$SdkHarnessDoFnRunner.finishBundle(ExecutableStageDoFnOperator.java:667) >>>>>>>> at >>>>>>>> org.apache.beam.runners.core.StatefulDoFnRunner.finishBundle(StatefulDoFnRunner.java:144) >>>>>>>> at >>>>>>>> org.apache.beam.runners.flink.translation.wrappers.streaming.ExecutableStageDoFnOperator$2.finishBundle(ExecutableStageDoFnOperator.java:754) >>>>>>>> at >>>>>>>> org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.finishBundle(DoFnRunnerWithMetricsUpdate.java:86) >>>>>>>> at >>>>>>>> org.apache.beam.runners.core.SimplePushbackSideInputDoFnRunner.finishBundle(SimplePushbackSideInputDoFnRunner.java:118) >>>>>>>> at >>>>>>>> org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.invokeFinishBundle(DoFnOperator.java:750) >>>>>>>> at >>>>>>>> org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.checkInvokeFinishBundleByTime(DoFnOperator.java:744) >>>>>>>> at >>>>>>>> org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.lambda$open$1(DoFnOperator.java:460) >>>>>>>> at >>>>>>>> org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$RepeatedTriggerTask.run(SystemProcessingTimeService.java:330) >>>>>>>> ... 7 more >>>>>>>> Caused by: java.util.concurrent.ExecutionException: >>>>>>>> java.lang.RuntimeException: Error received from SDK harness for >>>>>>>> instruction >>>>>>>> 6: Traceback (most recent call last): >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", >>>>>>>> line 307, in get >>>>>>>> processor = >>>>>>>> self.cached_bundle_processors[bundle_descriptor_id].pop() >>>>>>>> IndexError: pop from empty list >>>>>>>> >>>>>>>> During handling of the above exception, another exception occurred: >>>>>>>> >>>>>>>> Traceback (most recent call last): >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/internal/pickler.py", >>>>>>>> line 261, in loads >>>>>>>> return dill.loads(s) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/dill/_dill.py", >>>>>>>> line 317, in loads >>>>>>>> return load(file, ignore) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/dill/_dill.py", >>>>>>>> line 305, in load >>>>>>>> obj = pik.load() >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/dill/_dill.py", >>>>>>>> line 474, in find_class >>>>>>>> return StockUnpickler.find_class(self, module, name) >>>>>>>> *AttributeError: Can't get attribute '_timestamp_keyed_result' on >>>>>>>> <module 'pricingrealtime.aggregation.aggregation_transform' from >>>>>>>> '/srv/testpricingrt/7e5425fc4400bf14e80a98fb13461530c7867891/pricingrealtime/aggregation/aggregation_transform.py'>* >>>>>>>> >>>>>>>> During handling of the above exception, another exception occurred: >>>>>>>> >>>>>>>> Traceback (most recent call last): >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", >>>>>>>> line 165, in _execute >>>>>>>> response = task() >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", >>>>>>>> line 198, in <lambda> >>>>>>>> self._execute(lambda: worker.do_instruction(work), work) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", >>>>>>>> line 351, in do_instruction >>>>>>>> request.instruction_id) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", >>>>>>>> line 371, in process_bundle >>>>>>>> instruction_id, request.process_bundle_descriptor_reference) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/sdk_worker.py", >>>>>>>> line 313, in get >>>>>>>> self.data_channel_factory) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 576, in __init__ >>>>>>>> self.ops = >>>>>>>> self.create_execution_tree(self.process_bundle_descriptor) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 620, in create_execution_tree >>>>>>>> descriptor.transforms, key=topological_height, reverse=True)]) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 619, in <listcomp> >>>>>>>> for transform_id in sorted( >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 544, in wrapper >>>>>>>> result = cache[args] = func(*args) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 603, in get_operation >>>>>>>> in descriptor.transforms[transform_id].outputs.items() >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 602, in <dictcomp> >>>>>>>> for tag, pcoll_id >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 601, in <listcomp> >>>>>>>> tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]] >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 544, in wrapper >>>>>>>> result = cache[args] = func(*args) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 603, in get_operation >>>>>>>> in descriptor.transforms[transform_id].outputs.items() >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 602, in <dictcomp> >>>>>>>> for tag, pcoll_id >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 601, in <listcomp> >>>>>>>> tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]] >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 544, in wrapper >>>>>>>> result = cache[args] = func(*args) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 606, in get_operation >>>>>>>> transform_id, transform_consumers) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 865, in create_operation >>>>>>>> return creator(self, transform_id, transform_proto, payload, >>>>>>>> consumers) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 1108, in create >>>>>>>> serialized_fn, parameter) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/runners/worker/bundle_processor.py", >>>>>>>> line 1146, in _create_pardo_operation >>>>>>>> dofn_data = pickler.loads(serialized_fn) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/apache_beam/internal/pickler.py", >>>>>>>> line 265, in loads >>>>>>>> return dill.loads(s) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/dill/_dill.py", >>>>>>>> line 317, in loads >>>>>>>> return load(file, ignore) >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/dill/_dill.py", >>>>>>>> line 305, in load >>>>>>>> obj = pik.load() >>>>>>>> File >>>>>>>> "/srv/venvs/service/trusty/service_venv_python3.6/lib/python3.6/site-packages/dill/_dill.py", >>>>>>>> line 474, in find_class >>>>>>>> return StockUnpickler.find_class(self, module, name) >>>>>>>> AttributeError: Can't get attribute '_timestamp_keyed_result' on >>>>>>>> <module 'pricingrealtime.aggregation.aggregation_transform' from >>>>>>>> '/srv/testpricingrt/7e5425fc4400bf14e80a98fb13461530c7867891/pricingrealtime/aggregation/aggregation_transform.py'> >>>>>>>> >>>>>>>>
