coiled / blog-notebooks Goto Github PK

distributed.protocol.core - CRITICAL - Failed to deserialize
Traceback (most recent call last):
  File "/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/protocol/core.py", line 105, in loads
    return msgpack.loads(
  File "msgpack/_unpacker.pyx", line 195, in msgpack._cmsgpack.unpackb
  File "/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/protocol/core.py", line 96, in _decode_default
    sub_frames = decompress(sub_header, sub_frames)
  File "/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/protocol/compression.py", line 225, in decompress
    return [
  File "/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/protocol/compression.py", line 226, in <listcomp>
    compressions[c]["decompress"](frame)
KeyError: 'blosc'
---------------------------------------------------------------------------
KeyError                                  Traceback (most recent call last)
<ipython-input-28-fd710caa12ed> in <module>
      1 home1 = '5f7664a6434235066df7c37b'
      2 
----> 3 home1_heating = df.loc[df.household == home1, 'electricity_heating_kwh'].compute()

/opt/conda/envs/coiled/lib/python3.9/site-packages/dask/base.py in compute(self, **kwargs)
    283         dask.base.compute
    284         """
--> 285         (result,) = compute(self, traverse=False, **kwargs)
    286         return result
    287 

/opt/conda/envs/coiled/lib/python3.9/site-packages/dask/base.py in compute(*args, **kwargs)
    565         postcomputes.append(x.__dask_postcompute__())
    566 
--> 567     results = schedule(dsk, keys, **kwargs)
    568     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    569 

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/client.py in get(self, dsk, keys, workers, allow_other_workers, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, actors, **kwargs)
   2705                     should_rejoin = False
   2706             try:
-> 2707                 results = self.gather(packed, asynchronous=asynchronous, direct=direct)
   2708             finally:
   2709                 for f in futures.values():

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/client.py in gather(self, futures, errors, direct, asynchronous)
   2019             else:
   2020                 local_worker = None
-> 2021             return self.sync(
   2022                 self._gather,
   2023                 futures,

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
    860             return future
    861         else:
--> 862             return sync(
    863                 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
    864             )

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
    336     if error[0]:
    337         typ, exc, tb = error[0]
--> 338         raise exc.with_traceback(tb)
    339     else:
    340         return result[0]

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/utils.py in f()
    319             if callback_timeout is not None:
    320                 future = asyncio.wait_for(future, callback_timeout)
--> 321             result[0] = yield future
    322         except Exception:
    323             error[0] = sys.exc_info()

/opt/conda/envs/coiled/lib/python3.9/site-packages/tornado/gen.py in run(self)
    760 
    761                     try:
--> 762                         value = future.result()
    763                     except Exception:
    764                         exc_info = sys.exc_info()

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/client.py in _gather(self, futures, errors, direct, local_worker)
   1913                 else:
   1914                     self._gather_future = future
-> 1915                 response = await future
   1916 
   1917             if response["status"] == "error":

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/client.py in _gather_remote(self, direct, local_worker)
   1964 
   1965             else:  # ask scheduler to gather data for us
-> 1966                 response = await retry_operation(self.scheduler.gather, keys=keys)
   1967 
   1968         return response

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/utils_comm.py in retry_operation(coro, operation, *args, **kwargs)
    383         dask.config.get("distributed.comm.retry.delay.max"), default="s"
    384     )
--> 385     return await retry(
    386         partial(coro, *args, **kwargs),
    387         count=retry_count,

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/utils_comm.py in retry(coro, count, delay_min, delay_max, jitter_fraction, retry_on_exceptions, operation)
    368                 delay *= 1 + random.random() * jitter_fraction
    369             await asyncio.sleep(delay)
--> 370     return await coro()
    371 
    372 

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/core.py in send_recv_from_rpc(**kwargs)
    862             name, comm.name = comm.name, "ConnectionPool." + key
    863             try:
--> 864                 result = await send_recv(comm=comm, op=key, **kwargs)
    865             finally:
    866                 self.pool.reuse(self.addr, comm)

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/core.py in send_recv(comm, reply, serializers, deserializers, **kwargs)
    645         await comm.write(msg, serializers=serializers, on_error="raise")
    646         if reply:
--> 647             response = await comm.read(deserializers=deserializers)
    648         else:
    649             response = None

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/comm/tcp.py in read(self, deserializers)
    216                 frames = unpack_frames(frames)
    217 
--> 218                 msg = await from_frames(
    219                     frames,
    220                     deserialize=self.deserialize,

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/comm/utils.py in from_frames(frames, deserialize, deserializers, allow_offload)
     77         res = await offload(_from_frames)
     78     else:
---> 79         res = _from_frames()
     80 
     81     return res

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/comm/utils.py in _from_frames()
     60     def _from_frames():
     61         try:
---> 62             return protocol.loads(
     63                 frames, deserialize=deserialize, deserializers=deserializers
     64             )

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/protocol/core.py in loads(frames, deserialize, deserializers)
    103                 return msgpack_decode_default(obj)
    104 
--> 105         return msgpack.loads(
    106             frames[0], object_hook=_decode_default, use_list=False, **msgpack_opts
    107         )

msgpack/_unpacker.pyx in msgpack._cmsgpack.unpackb()

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/protocol/core.py in _decode_default(obj)
     94                 if deserialize:
     95                     if "compression" in sub_header:
---> 96                         sub_frames = decompress(sub_header, sub_frames)
     97                     return merge_and_deserialize(
     98                         sub_header, sub_frames, deserializers=deserializers

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/protocol/compression.py in decompress(header, frames)
    223 def decompress(header, frames):
    224     """Decompress frames according to information in the header"""
--> 225     return [
    226         compressions[c]["decompress"](frame)
    227         for c, frame in zip(header["compression"], frames)

/opt/conda/envs/coiled/lib/python3.9/site-packages/distributed/protocol/compression.py in <listcomp>(.0)
    224     """Decompress frames according to information in the header"""
    225     return [
--> 226         compressions[c]["decompress"](frame)
    227         for c, frame in zip(header["compression"], frames)
    228     ]

KeyError: 'blosc'