向DistributedArray同步并行输出向量时遇BoundsError问题求助

错误原因 #

distribute会将数组拆分到各个工作进程，每个进程的localpart(a)仅包含原数组的一部分元素。你在@distributed循环中使用全局索引i去访问本地数组，当i超出本地数组的长度时，就会触发BoundsError（比如例子中本地数组只有145个元素，但全局索引跑到了147）。

正确实现方式 #

方式一：按进程处理本地索引范围 #

直接让每个进程处理自己分到的索引区间，避免全局索引越界：

a = distribute([Float64[] for _ in 1:nrow(df)])
@sync @distributed for pid in workers()
    # 获取当前进程负责的全局索引范围
    local_idxs = localindices(a)
    # 取对应的数据框行
    local_data = @view df[local_idxs, :]
    # 获取当前进程的本地数组块
    local_a = localpart(a)
    
    # 遍历本地数据，填充结果
    for (idx, row) in enumerate(eachrow(local_data))
        append!(local_a[idx], foo(row))
    end
end

方式二：遍历分布式数组的块 #

利用eachchunk直接操作分布式数组的每个本地块：

a = distribute([Float64[] for _ in 1:nrow(df)])
@sync @distributed for chunk in eachchunk(a)
    local_a = chunk.data
    # 取当前块对应的全局索引的df行
    local_data = df[chunk.indices, :]
    
    for (i, row) in enumerate(eachrow(local_data))
        append!(local_a[i], foo(row))
    end
end

方式三：用pmap简化实现 #

如果不需要提前初始化分布式数组，pmap会自动分发任务并收集结果，代码更简洁：

# 并行计算每个行的结果
results = pmap(1:nrow(df)) do i
    foo(df[i])
end
# 若需要转为分布式数组
a = distribute(results)

注意事项 #

• 确保数据框df在所有工作进程中都能访问到。如果df体积较大，建议将其也转为分布式数组（如distribute(df)），减少跨进程数据传输开销。
• 若foo函数本身有副作用或依赖全局状态，需确保在每个进程中都正确初始化相关环境。

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