Error

[Mtkostecki]

Hello!

I have a problem when I try to use this piece of code:

%%time
varr = save_minian(
varr.chunk({"frame": chk["frame"], "height": -1, "width": -1}).rename("varr"),
intpath,
overwrite=True,
)

I have a video in .avi format, that is around 12GB.

In Jupyter notebook, I get:

---

KilledWorker Traceback (most recent call last)
File :1

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\minian\utilities.py:524, in save_minian(var, dpath, meta_dict, overwrite, chunks, compute, mem_limit)
522 except FileNotFoundError:
523 pass
--> 524 arr = ds.to_zarr(fp, compute=compute, mode=md)
525 if (chunks is not None) and compute:
526 chunks = {d: var.sizes[d] if v <= 0 else v for d, v in chunks.items()}

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\xarray\core\dataset.py:1745, in Dataset.to_zarr(self, store, chunk_store, mode, synchronizer, group, encoding, compute, consolidated, append_dim, region)
1742 if encoding is None:
1743 encoding = {}
-> 1745 return to_zarr(
1746 self,
1747 store=store,
1748 chunk_store=chunk_store,
1749 mode=mode,
1750 synchronizer=synchronizer,
1751 group=group,
1752 encoding=encoding,
1753 compute=compute,
1754 consolidated=consolidated,
1755 append_dim=append_dim,
1756 region=region,
1757 )

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\xarray\backends\api.py:1482, in to_zarr(dataset, store, chunk_store, mode, synchronizer, group, encoding, compute, consolidated, append_dim, region)
1480 # TODO: figure out how to properly handle unlimited_dims
1481 dump_to_store(dataset, zstore, writer, encoding=encoding)
-> 1482 writes = writer.sync(compute=compute)
1484 if compute:
1485 _finalize_store(writes, zstore)

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\xarray\backends\common.py:155, in ArrayWriter.sync(self, compute)
149 import dask.array as da
151 # TODO: consider wrapping targets with dask.delayed, if this makes
152 # for any discernable difference in perforance, e.g.,
153 # targets = [dask.delayed(t) for t in self.targets]
--> 155 delayed_store = da.store(
156 self.sources,
157 self.targets,
158 lock=self.lock,
159 compute=compute,
160 flush=True,
161 regions=self.regions,
162 )
163 self.sources = []
164 self.targets = []

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\dask\array\core.py:1022, in store(sources, targets, lock, regions, compute, return_stored, **kwargs)
1019 result = Delayed(name, dsk)
1021 if compute:
-> 1022 result.compute(**kwargs)
1023 return None
1024 else:

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\dask\base.py:281, in DaskMethodsMixin.compute(self, **kwargs)
257 def compute(self, **kwargs):
258 """Compute this dask collection
259
260 This turns a lazy Dask collection into its in-memory equivalent.
(...)
279 dask.base.compute
280 """
--> 281 (result,) = compute(self, traverse=False, **kwargs)
282 return result

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\dask\base.py:563, in compute(*args, **kwargs)
560 keys.append(x.dask_keys())
561 postcomputes.append(x.dask_postcompute())
--> 563 results = schedule(dsk, keys, **kwargs)
564 return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\distributed\client.py:2655, in Client.get(self, dsk, keys, workers, allow_other_workers, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, actors, **kwargs)
2653 should_rejoin = False
2654 try:
-> 2655 results = self.gather(packed, asynchronous=asynchronous, direct=direct)
2656 finally:
2657 for f in futures.values():

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\distributed\client.py:1964, in Client.gather(self, futures, errors, direct, asynchronous)
1962 else:
1963 local_worker = None
-> 1964 return self.sync(
1965 self._gather,
1966 futures,
1967 errors=errors,
1968 direct=direct,
1969 local_worker=local_worker,
1970 asynchronous=asynchronous,
1971 )

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\distributed\client.py:838, in Client.sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
836 return future
837 else:
--> 838 return sync(
839 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
840 )

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\distributed\utils.py:340, in sync(loop, func, callback_timeout, *args, **kwargs)
338 if error[0]:
339 typ, exc, tb = error[0]
--> 340 raise exc.with_traceback(tb)
341 else:
342 return result[0]

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\distributed\utils.py:324, in sync..f()
322 if callback_timeout is not None:
323 future = asyncio.wait_for(future, callback_timeout)
--> 324 result[0] = yield future
325 except Exception as exc:
326 error[0] = sys.exc_info()

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\tornado\gen.py:769, in Runner.run(self)
766 exc_info = None
768 try:
--> 769 value = future.result()
770 except Exception:
771 exc_info = sys.exc_info()

File ~\anaconda3\envs\minainWorkshop\lib\site-packages\distributed\client.py:1829, in Client._gather(self, futures, errors, direct, local_worker)
1827 exc = CancelledError(key)
1828 else:
-> 1829 raise exception.with_traceback(traceback)
1830 raise exc
1831 if errors == "skip":

KilledWorker: ("('load_avi_ffmpeg-from-value-f73b51143cce82a13fb82f74a0f84f97', 0, 0, 0)", <Worker 'tcp://127.0.0.1:59481', name: 3, memory: 0, processing: 1>)

On the console, I get:

av_interleaved_write_frame(): Broken pipekB time=00:04:29.96 bitrate=28049.8kbits/s speed=52.5x
Last message repeated 1 times
Error writing trailer of pipe:: Broken pipe
frame= 9016 fps=1564 q=-0.0 Lsize= 1028907kB time=00:05:00.53 bitrate=28046.2kbits/s speed=52.1x
video:1029039kB audio:0kB subtitle:0kB other streams:0kB global headers:0kB muxing overhead: unknown
Error closing file pipe:: Broken pipe
Conversion failed!