python - 带有 numpy 数组的 dask bag foldby

Question

当我在 上执行 foldby 时，我从 dask/numpy 收到了一条非常无用的 FutureWarning 消息>dask.bag 包含 numpy 数组。

def binop(a, b):
    print('binop')
    return a + b[1]

def combine(a, b):
    print('combine')
    return a + b[1]

seq = ((np.random.randint(0, 5, size=1)[0], np.ones(5,)) for _ in range(50))
db.from_sequence(seq, partition_size=10)\
    .foldby(0, binop=binop, initial=np.zeros(5,), combine=combine)\
    .compute()

目标只是将一堆 NumPy 数组相加。这会产生正确的结果，但也会从 NumPy 产生许多 FutureWarning 消息(看起来每个分区一个)，尽管它们看起来好像来自 dask.

dask/async.py:247: FutureWarning: elementwise comparison failed; returning scalar instead, but in the future will perform elementwise comparison return func(*args2)

仅将两个 numpy 数组相加而不使用 dask 不会产生结果，因此显然这里涉及并行 .foldby。看起来警告是在任何计算完成之前产生的。

• 如何确定警告是否是我应该关注的问题？
• 如果我应该担心警告，我该如何让警告消失？

我正在使用 python 3.6 dask 0.14.1 和 numpy 1.12.1

dask.bag.foldby

---

更新

感谢@MRocklin 的回答，我开始对此进行更多研究。所以 dask.async.py 中的违规代码是 this

def _execute_task(arg, cache, dsk=None):
....
    if isinstance(arg, list):
        return [_execute_task(a, cache) for a in arg]
    elif istask(arg):
        func, args = arg[0], arg[1:]
        args2 = [_execute_task(a, cache) for a in args]
        return func(*args2)

dask 是否有可能实际上试图遍历 args2 = [_execute_task(a, cache) for a in args] 中的 numpy 数组，我对内部的了解还不够(事实上)，无法判断这些变量包含什么。

Answer 1

这个警告确实来自 numpy。快速搜索代码库会得到 these lines :

    if (!strcmp(ufunc_name, "equal") ||
            !strcmp(ufunc_name, "not_equal")) {
        /* Warn on non-scalar, return NotImplemented regardless */
        assert(nin == 2);
        if (PyArray_NDIM(out_op[0]) != 0 ||
                PyArray_NDIM(out_op[1]) != 0) {
            if (DEPRECATE_FUTUREWARNING(
                    "elementwise comparison failed; returning scalar "
                    "instead, but in the future will perform elementwise "
                    "comparison") < 0) {
                return -1;
            }
        }

Dask 可能会使情况变得更糟，因为您会在每个进程中收到一次警告(dask.bag 默认使用进程池)。

此外，如果您的计算受 numpy 约束，那么您可能会考虑切换到线程调度程序而不是多进程调度程序

mybag.compute(get=dask.threaded.get)

参见 http://dask.pydata.org/en/latest/scheduler-choice.html