Hacktoberfest 2026：维护者为十月标记出来的 issue，仍然开放、适合新手。 浏览 Hacktoberfest issue

Exception when executing task graph with HighLevelGraph after a shuffle

未关闭
  • 哪里不对？
#1,129 3 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

还没有人认领这个 Issue。

评估

难度
4/5
预计耗时
3-5 天
新手友好度
42/100
Issue 类型
缺陷
描述清晰度
基本清楚
活跃度
停滞
技术栈
python
领域
data-engineering, distributed-systems

调研方向

使用 Dask 2024.8.1 和 dask-expr 1.1.11 运行最小的完整可验证示例，然后检查 dask_geopandas/sjoin.py 中的 HighLevelGraph 构建以及 dask_expr.from_graph 调用。跟踪 shuffle 键在 x1.compute() 期间的解析方式；当图将其替换为具体的 DataFrames 且示例在没有 TypeError 的情况下完成时，即表示完成。

由索引模型根据 Issue 内容生成。

描述

Describe the issue:

In https://github.com/geopandas/dask-geopandas/issues/303, dask-geopandas has a report that its custom sjoin method fails with a TypeError under some conditions. Internally, that method constructs a HighLevelGraph.

The observed failure is a TypeError raised by geopandas because dask fails to substitute the concrete (geo)DataFrame for the key (name, partition_number) when executing the task graph.

I've managed to produce a dask / dask-expr only version:

Minimal Complete Verifiable Example:

import dask.dataframe as dd
import pandas as pd
import dask
import dask_expr

dask.config.set(scheduler="single-threaded")


l1 = dd.from_pandas(pd.DataFrame({"a": [1, 2], "b": [0, 0]}), npartitions=1).shuffle("a")
r1 = dd.from_pandas(pd.DataFrame({"a": [1, 3], "c": [1, 1]}), npartitions=1).shuffle("a")


def func(left, right):
    assert isinstance(left, pd.DataFrame), f"Wrong type {left}"
    return pd.merge(left, right, how="inner")


dsk = {}
name = "myjoin-" + dask.base.tokenize(l1, r1, ensure_deterministic=True)
for i in range(1):
    dsk[(name, i)] = (
        func,
        (l1._name, i),
        (r1._name, i),
    )

divisions = [None] * (len(dsk) + 1)
g1 = dask.highlevelgraph.HighLevelGraph.from_collections(name, dsk, dependencies=[l1, r1])
x1 = dask_expr.from_graph(g1, func(l1._meta, r1._meta), divisions, dsk.keys(), "myjoin")
x1.compute()

That fails with

---------------------------------------------------------------------------
AssertionError                            Traceback (most recent call last)
Cell In[3], line 30
     28 g1 = dask.highlevelgraph.HighLevelGraph.from_collections(name, dsk, dependencies=[l1, r1])
     29 x1 = dask_expr.from_graph(g1, func(l1._meta, r1._meta), divisions, dsk.keys(), "myjoin")
---> 30 x1.compute()

File [~/gh/geopandas/dask-geopandas/.direnv/python-3.10/lib/python3.10/site-packages/dask_expr/_collection.py:477](http://127.0.0.1:8889/~/gh/geopandas/dask-geopandas/.direnv/python-3.10/lib/python3.10/site-packages/dask_expr/_collection.py#line=476), in FrameBase.compute(self, fuse, **kwargs)
    475     out = out.repartition(npartitions=1)
    476 out = out.optimize(fuse=fuse)
--> 477 return DaskMethodsMixin.compute(out, **kwargs)

File [~/gh/geopandas/dask-geopandas/.direnv/python-3.10/lib/python3.10/site-packages/dask/base.py:376](http://127.0.0.1:8889/~/gh/geopandas/dask-geopandas/.direnv/python-3.10/lib/python3.10/site-packages/dask/base.py#line=375), in DaskMethodsMixin.compute(self, **kwargs)
    352 def compute(self, **kwargs):
    353     """Compute this dask collection
    354 
    355     This turns a lazy Dask collection into its in-memory equivalent.
   (...)
    374     dask.compute
    375     """
--> 376     (result,) = compute(self, traverse=False, **kwargs)
    377     return result

File [~/gh/geopandas/dask-geopandas/.direnv/python-3.10/lib/python3.10/site-packages/dask/base.py:662](http://127.0.0.1:8889/~/gh/geopandas/dask-geopandas/.direnv/python-3.10/lib/python3.10/site-packages/dask/base.py#line=661), in compute(traverse, optimize_graph, scheduler, get, *args, **kwargs)
    659     postcomputes.append(x.__dask_postcompute__())
    661 with shorten_traceback():
--> 662     results = schedule(dsk, keys, **kwargs)
    664 return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])

Cell In[3], line 14, in func(left, right)
     13 def func(left, right):
---> 14     assert isinstance(left, pd.DataFrame), f"Wrong type {left}"
     15     return pd.merge(left, right, how="inner")

AssertionError: Wrong type ('rearrangebycolumn-8e5135a4ed46029cf9bb3c4bc7eff32e', 0)

As I mentioned, the shuffle there is important. Without that shuffle, things work fine.

Anything else we need to know?:

I'll take a look at this today.

Environment:

  • Dask version: '2024.8.1'
  • dask-expr version: '1.1.11'
主要语言
Python
星标
89
派生
26
30 天内没有已合并 PR

环境准备

  • 没有 Dockerfile 或 Docker Compose 文件
  • 没有 Pull Request 模板
  • 阅读贡献指南

从这里开始

  1. 先读完整个 Issue，再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库，在一个分支上完成修改。
  4. 提交 Pull Request，并在描述里引用这个 Issue 编号。

dask/dask-expr 的其他 Issue

查看 dask/dask-expr 的全部 Issue

相似的 Issue

更多 Python Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。