Hi,
When I run a task locally using `flow.run()` ,...
# ask-communitys
Shriram Holla
05/30/2022, 2:38 PMHi,
When I run a task locally using
flow.run()flow.register() Copy code
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 736, in save_tuple
save(element)
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 847, in _batch_setitems
save(v)
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 476, in save
f(self, obj) # Call unbound method with explicit self
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 821, in save_dict
self._batch_setitems(obj.items())
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 847, in _batch_setitems
save(v)
File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/pickle.py", line 496, in save
rv = reduce(self.proto)
File "/Users/shriramholla/.pyenvs/python3.6/lib/python3.6/site-packages/absl/flags/_flagvalues.py", line 677, in __getstate__
raise TypeError("can't pickle FlagValues")
TypeError: can't pickle FlagValuesa
Anna Geller
05/30/2022, 2:43 PM
Where did you get this syntax from? Prefect 1.0 has flow.register() - no need to register tasks separately
s
Shriram Holla
05/30/2022, 2:47 PMOops sorry I meant
flow.registera
Anna Geller
05/30/2022, 2:50 PM
this will answer your question - LMK if not https://docs.prefect.io/orchestration/flow_config/storage.html#pickle-vs-script-based-storage
s
Shriram Holla
05/30/2022, 3:35 PMIf I use a script based flow storage, will it allow me to run concurrent tasks?
a
Anna Geller
05/30/2022, 3:53 PM
Sure, yes - storage is independent of the executor (in 2.0 task runner)
k
Kevin Kho
05/30/2022, 4:20 PM
You are just returning something in a task that can’t be pickled. You can turn off checkpointing for the task too:
Copy code
@task(checkpoint=False)
def mytask():
return FlagValuess
Shriram Holla
05/30/2022, 4:23 PMI tried that but it didn’t work. That’s the fishy part because I’m not returning anything from any task.
k
Kevin Kho
05/30/2022, 4:27 PM
Ah sorry. Anna is completely right. If you have a global variable that is used by your tasks, it gets serialized along with the Flow
s
Shriram Holla
05/30/2022, 4:31 PMI see
Shriram Holla
05/30/2022, 4:37 PMSo here’s where I’m at right now.
I can run a flow locally but cannot register it due to the above issue.
I fixed this by using a script based flow storage. However, that requires a python script as an entrypoint. But since that would mean that it doesn’t run through our build system, I get import errors.
The build system we use generates modules during compilation and converts them into executables.
Do you know if there’s an alternate solution?
a
Anna Geller
05/30/2022, 4:39 PM
you can avoid that by switching to script storage - the docs I shared shows that
if you register via CLI, it uses script storage by default:
Copy code
prefect register --project xyz -p yourflow.pys
Shriram Holla
05/30/2022, 4:39 PMYeah that’s what I used, which ended up giving me import errors.
Shriram Holla
05/30/2022, 4:42 PMEssentially, do you know if there’s a way to do
Copy code
bazel run <register_flow_script>a
Anna Geller
05/30/2022, 5:36 PM
I don't know bazel :) but you would need Prefect CLI to trigger Prefect action and bazel CLI to trigger bazel action
54 Views