https://prefect.io logo
#prefect-community
Title
# prefect-community
m

Manuel Mourato

05/07/2020, 6:08 PM
I realize this a long shot, but maybe someone has had this issue before: I want to save a flow with a custom Task that I created. This custom task implements some logic with Cython, and when I execute the flow.save method, I get:
Copy code
rv = reduce(self.proto)
File "stringsource", line 2, in jnius.JavaClass.__reduce_cython__
TypeError: no default __reduce__ due to non-trivial __cinit__
A Cython object as far as I was able to search cannot be pickeled, unless specific code is add like so : https://stackoverflow.com/questions/12646436/pickle-cython-class My goal here, is to store this flow so that I can run an agent which will then execute it when I run via the UI. If I just register the flow in the UI , but dont save it, I get this error:
Copy code
[2020-05-07 17:43:48,846] INFO - agent |     flow = storage.get_flow(storage.flows[flow_data.name])
[2020-05-07 17:43:48,846] INFO - agent | KeyError: 'Test-Flow55'
My question is: did someone have an issue like this before? Can I get around this in some way?
👀 1
j

Jenny

05/07/2020, 6:32 PM
Hi @Manuel Mourato. Let me look into this one for you and see if I can get any ideas.
m

Manuel Mourato

05/07/2020, 6:33 PM
Thank you so much @Jenny, if you need anything tell me
j

Jenny

05/07/2020, 6:57 PM
Looks like there are two issues: Trying to save a flow that can't be pickled and a KeyError from registering and running that flow. Can you try using a different flow name and registering and running the flow again (without trying to save it) and see what happens?
m

Manuel Mourato

05/07/2020, 7:06 PM
Hey @Jenny, if I register the flow without trying to save, it appears in the UI But if I try to make a new run of this flow, I get the second error: KeyError: "Test-Flow66"
I assume because it cant find it in the storage
j

Jenny

05/07/2020, 7:07 PM
As background, KeyErrors usually come from a difference between the content Cloud expects and what's in the flow. But can you tell me more about why you think it might be a storage issue?
m

Manuel Mourato

05/07/2020, 7:09 PM
Because the method where I have this error, according to the log, is storage.get_flow(storage.flows.. )
j

Jenny

05/07/2020, 7:26 PM
Ok, let me see if we can get any more ideas
l

Laura Lorenz (she/her)

05/07/2020, 7:59 PM
Hi @Manuel Mourato, I’m sneaking in here too! I think we got a little confused on the part of calling
flow.save
directly, but in the end no matter how you slice it at some point your flow object will be pickled to be placed in storage if you want to use it with Cloud or Prefect Core’s server. Since the flow object itself must be pickleable, the tasks attached to them must be pickleable too, though they may produce unpickleable things during task run method since the run method is not executed until later. So, if you can defer the instantiation of whatever it is that is unpickleable until the task’s run method, then you’d be ok. (Which I think as you mention in this case is the Cython object, though tbh I don’t 100% understand the exception you’ve got)
j

Jenny

05/07/2020, 8:05 PM
Thanks @Laura Lorenz (she/her) - hope that helps @Manuel Mourato!
m

Manuel Mourato

05/07/2020, 9:47 PM
Thank you both for your help! For now I will put this issue on hold, and clearly it has nothing to do with Prefect's implementation, so no point discussing it further here
👍 2