I have had this issue with 2 flows now getting a flow state Prefect Community #ask-community

I have had this issue with 2 flows now, getting a ...

Mitchell Bregman

02/10/2021, 5:10 PM

I have had this issue with 2 flows now, getting a flow state of

Failed

on the UI, while every task and the flow itself is working as expected.. what can be causing this?

Copy code

[10 February 2021 12:06pm]: Unexpected error: TypeError('Could not serialize object of type Success.\nTraceback (most recent call last):\n File "/usr/local/lib/python3.7/site-packages/distributed/protocol/pickle.py", line 49, in dumps\n result = pickle.dumps(x, **dump_kwargs)...

josh

02/10/2021, 5:23 PM

Hi @Mitchell Bregman this error looks to me like you have a task returning some piece of data as the result that it is unable to pickle. Any chance you have more information surrounding this error being raised?

Mitchell Bregman

02/10/2021, 6:08 PM

@josh - do you want the full trace?

Mitchell Bregman

02/10/2021, 6:09 PM

here is the flow

Mitchell Bregman

02/10/2021, 6:09 PM

Copy code

with flow:
    alert_on = Parameter(
        "alert_on", default=["aws-prod-phonixxdw-5-7", "sf_dms_db", "aws_cape_rpt_new"]
    )
    slack_channel = Parameter("slack_channel", default="#iceburgers_comms")
    cred_metadata = tasks.looker_login.map(endpoints)
    connection_info = tasks.get_connections.map(cred_metadata)
    ping_list = tasks.reduce_connections(connection_info)
    pings = tasks.ping_connection.map(ping_list)
    alert = tasks.alert_failure.map(pings, unmapped(alert_on), unmapped(slack_channel))
    store = tasks.snowflake.map(data=alert)

Mitchell Bregman

02/10/2021, 6:09 PM

in case its useful

josh

02/10/2021, 6:09 PM

Yeah a full trace could help here

Mitchell Bregman

02/10/2021, 6:09 PM

copying from prefect UI

Mitchell Bregman

02/10/2021, 6:09 PM

Copy code

[10 February 2021 1:06pm]: Unexpected error: TypeError('Could not serialize object of type Success.\nTraceback (most recent call last):\n File "/usr/local/lib/python3.7/site-packages/distributed/protocol/pickle.py", line 49, in dumps\n result = pickle.dumps(x, **dump_kwargs)\nTypeError: can\'t pickle _thread.lock objects\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n File "/usr/local/lib/python3.7/site-packages/distributed/protocol/serialize.py", line 297, in serialize\n header, frames = dumps(x, context=context) if wants_context else dumps(x)\n File "/usr/local/lib/python3.7/site-packages/distributed/protocol/serialize.py", line 63, in pickle_dumps\n protocol=context.get("pickle-protocol", None) if context else None,\n File "/usr/local/lib/python3.7/site-packages/distributed/protocol/pickle.py", line 60, in dumps\n result = cloudpickle.dumps(x, **dump_kwargs)\n File "/usr/local/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 102, in dumps\n cp.dump(obj)\n File "/usr/local/lib/python3.7/site-packages/cloudpickle/cloudpickle_fast.py", line 563, in dump\n return Pickler.dump(self, obj)\n File "/usr/local/lib/python3.7/pickle.py", line 437, in dump\n self.save(obj)\n File "/usr/local/lib/python3.7/pickle.py", line 549, in save\n self.save_reduce(obj=obj, *rv)\n File "/usr/local/lib/python3.7/pickle.py", line 662, in save_reduce\n save(state)\n File "/usr/local/lib/python3.7/pickle.py", line 504, in save\n f(self, obj) # Call unbound method with explicit self\n File "/usr/local/lib/python3.7/pickle.py", line 859, in save_dict\n self._batch_setitems(obj.items())\n File "/usr/local/lib/python3.7/pickle.py", line 885, in _batch_setitems\n save(v)\n File "/usr/local/lib/python3.7/pickle.py", line 549, in save\n self.save_reduce(obj=obj, *rv)\n File "/usr/local/lib/python3.7/pickle.py", line 662, in save_reduce\n save(state)\n File "/usr/local/lib/python3.7/pickle.py", line 504, in save\n f(self, obj) # Call unbound method with explicit self\n File "/usr/local/lib/python3.7/pickle.py", line 859, in save_dict\n self._batch_setitems(obj.items())\n File "/usr/local/lib/python3.7/pickle.py", line 885, in _batch_setitems\n save(v)\n File "/usr/local/lib/python3.7/pickle.py", line 549, in save\n self.save_reduce(obj=obj, *rv)\n File "/usr/local/lib/python3.7/pickle.py", line 662, in save_reduce\n save(state)\n File "/usr/local/lib/python3.7/pickle.py", line 504, in save\n f(self, obj) # Call unbound method with explicit self\n File "/usr/local/lib/python3.7/pickle.py", line 859, in save_dict\n self._batch_setitems(obj.items())\n File "/usr/local/lib/python3.7/pickle.py", line 885, in _batch_setitems\n save(v)\n File "/usr/local/lib/python3.7/pickle.py", line 549, in save\n self.save_reduce(obj=obj, *rv)\n File "/usr/local/lib/python3.7/pickle.py", line 662, in save_reduce\n save(state)\n File "/usr/local/lib/python3.7/pickle.py", line 504, in save\n f(self, obj) # Call unbound method with explicit self\n File "/usr/local/lib/python3.7/pickle.py", line 859, in save_dict\n self._batch_setitems(obj.items())\n File "/usr/local/lib/python3.7/pickle.py", line 885, in _batch_setitems\n save(v)\n File "/usr/local/lib/python3.7/pickle.py", line 549, in save\n self.save_reduce(obj=obj, *rv)\n File "/usr/local/lib/python3.7/pickle.py", line 662, in save_reduce\n save(state)\n File "/usr/local/lib/python3.7/pickle.py", line 504, in save\n f(self, obj) # Call unbound method with explicit self\n File "/usr/local/lib/python3.7/pickle.py", line 859, in save_dict\n self._batch_setitems(obj.items())\n File "/usr/local/lib/python3.7/pickle.py", line 885, in _batch_setitems\n save(v)\n File "/usr/local/lib/python3.7/pickle.py", line 524, in save\n rv = reduce(self.proto)\nTypeError: can\'t pickle _thread.lock objects\n')

Mitchell Bregman

02/10/2021, 6:10 PM

the last task is the SnowflakeQuery built-in

josh

02/10/2021, 6:11 PM

Ah so it looks like one of your tasks is returning a

_thread.lock

object which I have seen as an issue in the past. You said you are seeing it two flows, are there any tasks that they share? I wonder if there’s a chance that the SnowflakeQuery task could be the bug and if so we should open an issue report for it

Mitchell Bregman

02/10/2021, 6:17 PM

yes, we are both using the

SnowflakeQuery

Mitchell Bregman

02/10/2021, 6:17 PM

one of the flows is running an INSERT statement and the other is CALL stored proc

Mitchell Bregman

02/10/2021, 6:18 PM

i can build a custom SnowflakeQuery task but rather just use the built-in for convenience

josh

02/10/2021, 6:18 PM

I wonder if you just copy the SnowflakeQuery task source and adjust the return type if it’ll fix it 🤔

Mitchell Bregman

02/10/2021, 6:18 PM

and it is definitely the snowflake query because i just added that task and this is the first time im seeing this error… before, i was just alerting and not storing the info

Mitchell Bregman

02/10/2021, 6:19 PM

whats the current return type

josh

02/10/2021, 6:19 PM

Ah I found an issue where this is raised! https://github.com/PrefectHQ/prefect/issues/3744

josh

02/10/2021, 6:19 PM

So this is known

josh

02/10/2021, 6:20 PM

The current return type is a

cursor.execute

result from the snowflake connector library

Mitchell Bregman

02/10/2021, 6:20 PM

gotcha

Mitchell Bregman

02/10/2021, 6:21 PM

so interestingly, this also seems to tie into the DaskExecutor

Mitchell Bregman

02/10/2021, 6:21 PM

when running wth a LocalExecutor, no issues are found… no issues with LocalDaskExecutor either

josh

02/10/2021, 6:21 PM

Hm yeah I believe that the Dask executor requires that the results are pickleable in order to pass them around

Mitchell Bregman

02/10/2021, 6:22 PM

gotcha

Mitchell Bregman

02/10/2021, 6:22 PM

ok - keep me posted, ill follow that issue thread

Open in Slack

Previous Next