Dask error when reading data from a large zip file

Question

I have a .zip folder with a lot of .csv files which I read into a Dask DataFrame like this:

import dask.dataframe as dd

df = dd.read_csv(file_names, storage_options={'fo': folder_path})

Where each file_name.csv is prepended with a 'zip://' tag and folder_path is folder_name.zip.

When I then try to perform any operation such as

df.set_index('new_index')

or

df.groupby(['column']).mean().compute()

I often get the following error stack:

Traceback (most recent call last):
  File "<input>", line 5, in <module>
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\base.py", line 166, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\base.py", line 437, in compute
    results = schedule(dsk, keys, **kwargs)
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\threaded.py", line 84, in get
    **kwargs
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\local.py", line 486, in get_async
    raise_exception(exc, tb)
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\local.py", line 316, in reraise
    raise exc
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\local.py", line 222, in execute_task
    result = _execute_task(task, data)
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\core.py", line 121, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\core.py", line 121, in <genexpr>
    return func(*(_execute_task(a, cache) for a in args))
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\core.py", line 121, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\core.py", line 121, in <genexpr>
    return func(*(_execute_task(a, cache) for a in args))
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\core.py", line 121, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\core.py", line 121, in <genexpr>
    return func(*(_execute_task(a, cache) for a in args))
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\core.py", line 115, in _execute_task
    return [_execute_task(a, cache) for a in arg]
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\core.py", line 115, in <listcomp>
    return [_execute_task(a, cache) for a in arg]
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\core.py", line 121, in _execute_task
    return func(*(_execute_task(a, cache) for a in args))
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\dask\bytes\core.py", line 184, in read_block_from_file
    return read_block(f, off, bs, delimiter)
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\site-packages\fsspec\utils.py", line 258, in read_block
    b = f.read(length)
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\zipfile.py", line 930, in read
    data = self._read1(n)
  File "C:\Users\Vlad\Miniconda3\envs\pvforecast\lib\zipfile.py", line 1006, in _read1
    data = self._decompressor.decompress(data, n)
zlib.error: Error -3 while decompressing data: invalid block type

The weirdest thing is that it sometimes works perfectly fine for the same dataset without changing any code. It also always takes a different amount of time to raise the error and does it for different parts of code. Seems like it could be a bug within dask, but I don't get why it works at one time, but not the other?