Channels
announcements
ask-marvin
best-practices-coordination-plane
data-ecosystem
data-tricks-and-tips
events
find-a-prefect-job
geo-australia
geo-bay-area
geo-berlin
geo-boston
geo-chicago
geo-colorado
geo-dc
geo-israel
geo-japan
geo-london
geo-nyc
geo-seattle
geo-texas
gratitude
introductions
marvin-in-the-wild
prefect-ai
prefect-aws
prefect-azure
prefect-cloud
prefect-community
prefect-contributors
prefect-dbt
prefect-docker
prefect-gcp
prefect-getting-started
prefect-integrations
prefect-kubernetes
prefect-recipes
prefect-server
prefect-ui
random
show-us-what-you-got
Title
g
Greg Desmarais
07/23/2020, 3:31 PMHas anyone seen an issue with Dask workers having errors around serialization? I'm submitting a flow from a client (simple hello world) -> prefect server on an ec2 instance -> prefect agent -> Dask scheduler -> Dask workers. Everything is flowing allthe way to the worker, but then I am seeing a
TypeError: can not serialize 'Variable' object==> prefect_logs/dask_worker.log <==
distributed.protocol.core - CRITICAL - Failed to Serialize
Traceback (most recent call last):
File "/opt/conda/lib/python3.7/site-packages/distributed/protocol/core.py", line 25, in dumps
small_header, small_payload = dumps_msgpack(msg)
File "/opt/conda/lib/python3.7/site-packages/distributed/protocol/core.py", line 163, in dumps_msgpack
payload = msgpack.dumps(msg, use_bin_type=True)
File "/opt/conda/lib/python3.7/site-packages/msgpack/__init__.py", line 35, in packb
return Packer(**kwargs).pack(o)
File "msgpack/_packer.pyx", line 286, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 292, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 289, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 225, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 283, in msgpack._cmsgpack.Packer._pack
TypeError: can not serialize 'Variable' object
distributed.comm.utils - INFO - Unserializable Message: {'op': 'variable_get', 'timeout': 0, 'name': <distributed.variable.Variable object at 0x7fec9eafaef0>, 'client': 'Client-worker-365168fe-ccf9-11ea-800b-0242ac110004', 'reply': True}
distributed.comm.utils - ERROR - can not serialize 'Variable' object
Traceback (most recent call last):
File "/opt/conda/lib/python3.7/site-packages/distributed/comm/utils.py", line 35, in _to_frames
msg, serializers=serializers, on_error=on_error, context=context
File "/opt/conda/lib/python3.7/site-packages/distributed/protocol/core.py", line 25, in dumps
small_header, small_payload = dumps_msgpack(msg)
File "/opt/conda/lib/python3.7/site-packages/distributed/protocol/core.py", line 163, in dumps_msgpack
payload = msgpack.dumps(msg, use_bin_type=True)
File "/opt/conda/lib/python3.7/site-packages/msgpack/__init__.py", line 35, in packb
return Packer(**kwargs).pack(o)
File "msgpack/_packer.pyx", line 286, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 292, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 289, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 225, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 283, in msgpack._cmsgpack.Packer._pack
TypeError: can not serialize 'Variable' objectd
Dylan
07/23/2020, 5:12 PMHey @Greg Desmarais! Would you mind sharing your flow?
I will see if I can help or direct you to our resident Dask expert who almost certainly can 😄
l
Luke
07/23/2020, 7:35 PMUnserializable Message: {'op': 'variable_get', 'timeout': 0, 'name': <distributed.variable.Variable object at 0x7fec9eafaef0>, 'client': 'Client-worker-365168fe-ccf9-11ea-800b-0242ac110004', 'reply': True}maybe just extract a string representation?
g
Greg Desmarais
07/23/2020, 7:42 PMFlow on the way...
@task
def say_hello():
return f'done'
with Flow("Dask ECS Test 2") as flow:
say_hello()Simple enough?
😉
Then,
bucket = 'celsius-temp-data'
key = 'datasciences/prefect_flows/dask_ecs_flow_test_2'
flow.storage = S3(bucket, key=key)
labels = ['size-small']
executor = DaskExecutor(address=f'{PREFECT_COMPOSE_HOST}:38786')
flow.environment = LocalEnvironment(executor=executor, labels=labels)
flow_id = flow.register(labels=labels)
p_client = Client()
ret = p_client.create_flow_run(flow_id=flow_id)
print(f'Created flow run: {ret}')👀 1
I had some fargate cruft from previous efforts...
Also - the stack trace above isn't from my code - it is inside the dask stuff. I don't sall any of those ops directly.
d
Dylan
07/23/2020, 7:51 PMLooks like this was fixed recently! What version of Prefect are you using?
g
Greg Desmarais
07/23/2020, 7:55 PMYou are probably going to laugh a bit. In between the writing of this original posting and now, my pycharm told me there was an update to prefect - I was on 0.12.4. I went through the change logs and found that exact issue. I am about to kick off a re-build of all my images and the like to go to 0.12.5.
In reading through the fix, I'm not sure how it addresses the issue, but there is a lot of magic going on, and it certainly is in the right arena. I'll upgrade then post back here.
d
Dylan
07/23/2020, 7:56 PMGreat! 😄
g
Greg Desmarais
07/23/2020, 7:58 PMI was pretty sure I had searched the issues on github, but maybe not. It also looks like this pr might not be tied to an issue?
Oh - and the pr refers to an issue reported on slack...which was me 😉 :snake eating tail:
😂 2
Ok - after upgrading all over the place, you can consider the case closed! Got a nice setup working now...Thanks!
d
Dylan
07/23/2020, 8:38 PMYou’re welcome! Glad it all worked out 😄