binding delayeds over ddfs throws Missing dependency
- Dominant language
- Python
- Stars
- 13.9k
- Forks
- 2k
- PR merge metrics
- No merged PRs in 30d
Description
**Describe the issue**:
Calling Dask `bind` over `delayed`'s that take a `ddf` as input throws "Missing Dependency" errors.
**Minimal Complete Verifiable Example**:
```python
import dask.dataframe as dd
from dask.delayed import delayed
from dask.graph_manipulation import bind
from dask.distributed import Client
import pandas as pd
pdf_1 = pd.DataFrame({"numbers": [1, 2, 3], "words": ["one", "two", "three"]})
ddf_1 = dd.from_pandas(pdf_1, npartitions=1)
pdf_2 = pd.DataFrame({"numbers": [4, 5, 6], "words": ["four", "five", "six"]})
ddf_2 = dd.from_pandas(pdf_2, npartitions=1)
def print_df(pdf: pd.DataFrame) -> None:
for i, row in pdf.iterrows():
print(f'{row["numbers"]}: {row["words"]}')
delayed_1 = delayed(print_df)(ddf_1)
delayed_2 = delayed(print_df)(ddf_2)
compute_in_order = bind(delayed_2, delayed_1)
compute_in_order.compute()
```
Throws:
```
FutureCancelledError Traceback (most recent call last)
Cell In[13], [line 1](vscode-notebook-cell:?execution_count=13&line=1)
----> [1](vscode-notebook-cell:?execution_count=13&line=1) compute_in_order.compute()
File /mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:373, in DaskMethodsMixin.compute(self, **kwargs)
[349](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:349) def compute(self, **kwargs):
[350](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:350) """Compute this dask collection
[351](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:351)
[352](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:352) This turns a lazy Dask collection into its in-memory equivalent.
(...)
[371](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:371) dask.compute
[372](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:372) """
--> [373](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:373) (result,) = compute(self, traverse=False, **kwargs)
[374](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:374) return result
File /mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:681, in compute(traverse, optimize_graph, scheduler, get, *args, **kwargs)
[678](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:678) expr = expr.optimize()
[679](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:679) keys = list(flatten(expr.__dask_keys__()))
--> [681](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:681) results = schedule(expr, keys, **kwargs)
[683](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/dask/base.py:683) return repack(results)
File /mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2416, in Client._gather(self, futures, errors, direct, local_worker)
[2414](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2414) exception = st.exception
[2415](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2415) traceback = st.traceback
-> [2416](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2416) raise exception.with_traceback(traceback)
[2417](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2417) if errors == "skip":
[2418](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2418) bad_keys.add(key)
FutureCancelledError: print_df-ff5e943d8c14c027d0ec9806863f1730 cancelled for reason: lost dependencies.
```
And then using Dask distributed leads to a `lost dependencies` error:
```
client = Client()
future = client.compute(compute_in_order)
client.gather(future)
```
```
FutureCancelledError Traceback (most recent call last)
Cell In[14], [line 3](vscode-notebook-cell:?execution_count=14&line=3)
[1](vscode-notebook-cell:?execution_count=14&line=1) client = Client()
[2](vscode-notebook-cell:?execution_count=14&line=2) future = client.compute(compute_in_order)
----> [3](vscode-notebook-cell:?execution_count=14&line=3) client.gather(future)
File /mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2555, in Client.gather(self, futures, errors, direct, asynchronous)
[2552](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2552) local_worker = None
[2554](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2554) with shorten_traceback():
-> [2555](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2555) return self.sync(
[2556](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2556) self._gather,
[2557](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2557) futures,
[2558](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2558) errors=errors,
[2559](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2559) direct=direct,
[2560](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2560) local_worker=local_worker,
[2561](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2561) asynchronous=asynchronous,
[2562](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2562) )
File /mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2416, in Client._gather(self, futures, errors, direct, local_worker)
[2414](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2414) exception = st.exception
[2415](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2415) traceback = st.traceback
-> [2416](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2416) raise exception.with_traceback(traceback)
[2417](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2417) if errors == "skip":
[2418](https://vscode-remote+ssh-002dremote-002bdev.vscode-resource.vscode-cdn.net/mnt/home/andoni.garcia/code/research/.venv/lib/python3.11/site-packages/distributed/client.py:2418) bad_keys.add(key)
FutureCancelledError: print_df-ff5e943d8c14c027d0ec9806863f1730 cancelled for reason: lost dependencies.
```
**Anything else we need to know?**:
This seems to have happened when upgrading from
```
dask = {version = "2024.8.0", extras = ["complete", "distributed"]}
dask-expr = "^1.1.2"
```
to
```
dask = {extras = ["complete", "distributed"], version = "^2025.9.1"}
```
And is related to https://github.com/dask/dask/issues/12055
**Environment**:
- Dask version: `2025.10.0`
- Python version: `3.11.9`
- Operating System: Ubuntu EC2
- Install method (conda, pip, source): poetry
Contributor guide
Assessment
This issue has not been assessed yet.