dask / dask/dask

binding delayeds over ddfs throws Missing dependency

Open
#12,127 2 comments 0 reactions 0 assignees View on GitHub
bug
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.