blocks = [block_from_header_bytes(h, b)
File "/shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/pandas.py", line 162, in block_from_header_bytes
values = pnp.deserialize(pnp.decompress(bytes, dtype), dtype,
File "/shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/numpy.py", line 159, in decompress
return decompress_text(bytes)
File "/home/krishnamurthy.sur/.local/lib/python3.9/site-packages/snappy/snappy.py", line 92, in uncompress
return _uncompress(data)
snappy.UncompressError: Error while decompressing: invalid input
```
Code:
```auto
df = dd.read_csv('/work/scdatahub/Data_Cleaning/Output_Data/importyeti_apparel_indexed.csv/*',
dtype=dtype, encoding="utf-8")
addr = dd.read_csv('/work/scdatahub/Data_Cleaning/Input_Data/address.csv', encoding="utf-8")
...
merged = dd.merge(df, addr.add_prefix('con_'),
how='left',
left_on='prepped_consignee_addr', right_on='con_low_addr')
merged = dd.merge(merged, addr.add_prefix('ship_'),
how='left',
left_on='prepped_shipper_addr', right_on='ship_low_addr')
merged = merged.astype('string')
merged.to_csv('/work/scdatahub/Data_Cleaning/Output_Data/apparel_geocoded.csv')
```
---
### Author:  [@guillaumeeb](https://dask.discourse.group/u/guillaumeeb)
#### Post date: [April 7, 2023, 5:23am UTC](https://dask.discourse.group/t/error-while-decompressing-invalid-input/1743/2 "2023-04-07T05:23:28Z")
Hi @suryak, welcome to Dask community!
First, are you using Dask Distributed? What kind of Scheduler/Cluster Manager are you using?
Merge implies shuffling the data between workers, and it seems Workers or Scheduler aren’t speeking the smae language here when exanging data…
It’s a little hard to tell, but I did notice something: you seem to have some libraries installed on your local environment:
> [@suryak](#):
>
> `File "/home/krishnamurthy.sur/.local/lib/python3.9/site-packages/snappy/snappy.py"`
Maybe this local environment is not compatible with the main one in
> [@suryak](#):
>
> `/shared/centos7/anaconda3/2022.05`
You should try with a simple threaded Scheduler to see if it fixes the problem, and then probably check your environment.
---
### Author:  [@suryak](https://dask.discourse.group/u/suryak)
#### Post date: [April 7, 2023, 5:26am UTC](https://dask.discourse.group/t/error-while-decompressing-invalid-input/1743/3 "2023-04-07T05:26:40Z")
That’s a good catch! I’m using a slurm manager with my university’s HPC. I submit my job with the `sbatch` command. I will give my environment a closer look. Thank you!
---
### Author:  [@suryak](https://dask.discourse.group/u/suryak)
#### Post date: [April 8, 2023, 11:27pm UTC](https://dask.discourse.group/t/error-while-decompressing-invalid-input/1743/4 "2023-04-08T23:27:29Z")
I fixed the libraries so they’re all being used from the same environment but the error persists:
```auto
---------------------------------------------------------------------------
UncompressError Traceback (most recent call last)
File :1, in
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/dask/dataframe/core.py:1560, in _Frame.to_csv(self, filename, **kwargs)
1557 """See dd.to_csv docstring for more information"""
1558 from .io import to_csv
-> 1560 return to_csv(self, filename, **kwargs)
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/dask/dataframe/io/csv.py:957, in to_csv(df, filename, single_file, encoding, mode, name_function, compression, compute, scheduler, storage_options, header_first_partition_only, compute_kwargs, **kwargs)
953 compute_kwargs["scheduler"] = scheduler
955 import dask
--> 957 return list(dask.compute(*values, **compute_kwargs))
958 else:
959 return values
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/dask/base.py:573, in compute(traverse, optimize_graph, scheduler, get, *args, **kwargs)
570 keys.append(x. __dask_keys__ ())
571 postcomputes.append(x. __dask_postcompute__ ())
--> 573 results = schedule(dsk, keys, **kwargs)
574 return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/dask/threaded.py:81, in get(dsk, result, cache, num_workers, pool, **kwargs)
78 elif isinstance(pool, multiprocessing.pool.Pool):
79 pool = MultiprocessingPoolExecutor(pool)
---> 81 results = get_async(
82 pool.submit,
83 pool._max_workers,
84 dsk,
85 result,
86 cache=cache,
87 get_id=_thread_get_id,
88 pack_exception=pack_exception,
89 **kwargs,
90 )
92 # Cleanup pools associated to dead threads
93 with pools_lock:
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/dask/local.py:506, in get_async(submit, num_workers, dsk, result, cache, get_id, rerun_exceptions_locally, pack_exception, raise_exception, callbacks, dumps, loads, chunksize, **kwargs)
504 _execute_task(task, data) # Re-execute locally
505 else:
--> 506 raise_exception(exc, tb)
507 res, worker_id = loads(res_info)
508 state["cache"][key] = res
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/dask/local.py:314, in reraise(exc, tb)
312 if exc. __traceback__ is not tb:
313 raise exc.with_traceback(tb)
--> 314 raise exc
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/dask/local.py:219, in execute_task(key, task_info, dumps, loads, get_id, pack_exception)
217 try:
218 task, data = loads(task_info)
--> 219 result = _execute_task(task, data)
220 id = get_id()
221 result = dumps((result, id))
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/dask/core.py:119, in _execute_task(arg, cache, dsk)
115 func, args = arg[0], arg[1:]
116 # Note: Don't assign the subtask results to a variable. numpy detects
117 # temporaries by their reference count and can execute certain
118 # operations in-place.
--> 119 return func(*(_execute_task(a, cache) for a in args))
120 elif not ishashable(arg):
121 return arg
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/dask/dataframe/shuffle.py:815, in collect(p, part, meta, barrier_token)
813 """Collect partitions from partd, yield dataframes"""
814 with ensure_cleanup_on_exception(p):
--> 815 res = p.get(part)
816 return res if len(res) > 0 else meta
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/core.py:73, in Interface.get(self, keys, **kwargs)
71 def get(self, keys, **kwargs):
72 if not isinstance(keys, list):
---> 73 return self.get([keys], **kwargs)[0]
74 elif any(isinstance(key, list) for key in keys): # nested case
75 flatkeys = list(flatten(keys))
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/core.py:79, in Interface.get(self, keys, **kwargs)
77 return nested_get(keys, dict(zip(flatkeys, result)))
78 else:
---> 79 return self._get(keys, **kwargs)
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/encode.py:29, in Encode._get(self, keys, **kwargs)
27 def _get(self, keys, **kwargs):
28 raw = self.partd._get(keys, **kwargs)
---> 29 return [self.join([self.decode(frame) for frame in framesplit(chunk)])
30 for chunk in raw]
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/encode.py:29, in (.0)
27 def _get(self, keys, **kwargs):
28 raw = self.partd._get(keys, **kwargs)
---> 29 return [self.join([self.decode(frame) for frame in framesplit(chunk)])
30 for chunk in raw]
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/encode.py:29, in (.0)
27 def _get(self, keys, **kwargs):
28 raw = self.partd._get(keys, **kwargs)
---> 29 return [self.join([self.decode(frame) for frame in framesplit(chunk)])
30 for chunk in raw]
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/pandas.py:201, in deserialize(bytes)
198 bytes = frames[1:]
199 axes = [index_from_header_bytes(headers[0], bytes[0]),
200 index_from_header_bytes(headers[1], bytes[1])]
--> 201 blocks = [block_from_header_bytes(h, b)
202 for (h, b) in zip(headers[2:], bytes[2:])]
203 return pd.DataFrame(create_block_manager_from_blocks(blocks, axes))
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/pandas.py:201, in (.0)
198 bytes = frames[1:]
199 axes = [index_from_header_bytes(headers[0], bytes[0]),
200 index_from_header_bytes(headers[1], bytes[1])]
--> 201 blocks = [block_from_header_bytes(h, b)
202 for (h, b) in zip(headers[2:], bytes[2:])]
203 return pd.DataFrame(create_block_manager_from_blocks(blocks, axes))
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/pandas.py:162, in block_from_header_bytes(header, bytes)
160 values = pickle.loads(bytes)
161 else:
--> 162 values = pnp.deserialize(pnp.decompress(bytes, dtype), dtype,
163 copy=True).reshape(shape)
164 if extension_type == 'categorical_type':
165 values = pd.Categorical.from_codes(values,
166 extension_values[1],
167 ordered=extension_values[0])
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/partd/numpy.py:159, in decompress(bytes, dtype)
157 def decompress(bytes, dtype):
158 if dtype == 'O':
--> 159 return decompress_text(bytes)
160 else:
161 return decompress_bytes(bytes)
File /shared/centos7/anaconda3/2022.05/lib/python3.9/site-packages/snappy/snappy.py:92, in uncompress(data, decoding)
90 if decoding:
91 return _uncompress(data).decode(decoding)
---> 92 return _uncompress(data)
UncompressError: Error while decompressing: invalid input
```
I also set the config like so:
```auto
dask.config.set(scheduler='threads')
```
---
### Author:  [@guillaumeeb](https://dask.discourse.group/u/guillaumeeb)
#### Post date: [April 10, 2023, 5:38pm UTC](https://dask.discourse.group/t/error-while-decompressing-invalid-input/1743/5 "2023-04-10T17:38:35Z")
Well, this will be a little hard to investigate without a Minimum Reproducible Example. Do you think you could build one with part of your data, or fake ones? Ideally, an example using a `LocalCluster` so everyone can reproduce it on its laptop.