初始化Cluster和Client后，Dask DataFrames按索引合并报ValueError

在Dask 2023.6.0版本中，使用LocalCluster和Client时，合并两个共享索引的Dask DataFrame会触发ValueError: Length of values (0) does not match length of index (10)，但本地模式（不初始化集群）下可正常执行。

问题原因 #

分布式模式下，Dask为优化合并效率，会自动对两个DataFrame按索引执行哈希分区。但通过dd.from_delayed创建的DataFrame，Dask无法自动推断索引的分区范围信息，导致哈希分区过程中出现数据丢失，进而触发长度不匹配的错误。本地模式下Dask采用单进程执行策略，无需重新分区，因此不会出现该问题。

解决方案 #

方案1：显式设置索引分区 #

在创建DataFrame后，通过set_index明确将索引设为分区键，并指定与原始数据一致的分区数，让Dask获取索引的分区信息：

import dask.dataframe as dd
from dask.distributed import Client, LocalCluster
import pandas as pd
from dask import delayed

cluster = LocalCluster(n_workers=2)
client = Client(cluster)

def create_part(i, col):
    return pd.DataFrame(
        data={
            col: list(range(10)),
        },
        index=[f"{i}_{j}" for j in range(10)]
    )

df1 = dd.from_delayed([delayed(create_part)(i, "a") for i in range(3)])
df2 = dd.from_delayed([delayed(create_part)(i, "b") for i in range(3)])

# 显式设置索引分区，保持原始分区数
df1 = df1.set_index(df1.index, npartitions=3)
df2 = df2.set_index(df2.index, npartitions=3)

result = df1.merge(df2, left_index=True, right_index=True).compute()
print(result)

方案2：合并时指定任务级Shuffle #

在merge方法中设置shuffle='tasks'，强制Dask使用任务级别的Shuffle策略，避免自动哈希分区导致的错误：

import dask.dataframe as dd
from dask.distributed import Client, LocalCluster
import pandas as pd
from dask import delayed

cluster = LocalCluster(n_workers=2)
client = Client(cluster)

def create_part(i, col):
    return pd.DataFrame(
        data={
            col: list(range(10)),
        },
        index=[f"{i}_{j}" for j in range(10)]
    )

df1 = dd.from_delayed([delayed(create_part)(i, "a") for i in range(3)])
df2 = dd.from_delayed([delayed(create_part)(i, "b") for i in range(3)])

result = df1.merge(df2, left_index=True, right_index=True, shuffle='tasks').compute()
print(result)

内容的提问来源于stack exchange，提问作者Maitreya