Maurice Poirrier created BEAM-10384:
---------------------------------------
Summary: Dataflow can't pickle WeakRefDictionary
Key: BEAM-10384
URL: https://issues.apache.org/jira/browse/BEAM-10384
Project: Beam
Issue Type: Bug
Components: runner-dataflow
Affects Versions: 2.22.0
Environment: Ubuntu LTS (direct runner)
Reporter: Maurice Poirrier
Fix For: Not applicable
we are trying to deploy an Streaming pipeline to Dataflow where we separate in
few different "routes" that we manipulate differently the data.
We did the complete development with the `DirectRunner`, and works smoothly as
we tested but now, that we did deployed it to `Dataflow`, it does not work.
The code fails when yielding on the following `doFn`
``` python
class SplitByRoute(beam.DoFn):
OUTPUT_TAG_ROUTE_ONE= "route_one"
OUTPUT_TAG_ROUTE_TWO = "route_two"
OUTPUT_NOT_SUPPORTED = "not_supported"
def __init__(self):
beam.DoFn.__init__(self)
def process(self, elem):
try:
route = self.define_route(elem["param"]) # Just tag it depending on param
except Exception:
route = None
logging.info(f"Routed to \{route}")
if route == self.OUTPUT_TAG_ROUTE_ONE:
yield TaggedOutput(self.OUTPUT_TAG_ROUTE_ONE, elem)
elif route == self.OUTPUT_TAG_ROUTE_TWO:
logging.info(f"Element: \{elem}")
yield TaggedOutput(self.OUTPUT_TAG_ROUTE_TWO, elem)
else:
yield TaggedOutput(self.OUTPUT_NOT_SUPPORTED, elem)
```
It does log the element, yield the output and fails with the following error
`AttributeError: Can't pickle local object
'WeakValueDictionary.__init__.<locals>.remove' [while running
'generatedPtransform-3196']`
Other considerations are that we use taggedOutputs on the pipeline before this
`DoFn`, and it works on Dataflow but this one in particularly fails with the
error mentioned. Could it be the memory cache? or something related to it?
Where `Weakrefs` are used?
Far as I know, this error happens when you have a class inside another one.
Maybe not(?)
Any suggestions so how we could manage this? It's been very frustrating error.
Thank you!!! :)
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
