本文的目的是介绍使用Dask进行内存Numpy计算的工作线程与线程的详细情况，我们将通过专业的研究、有关数据的分析等多种方式，同时也不会遗漏关于AnacondaNumpy错误“Importingthe

本文的目的是介绍使用 Dask 进行内存 Numpy 计算的工作线程与线程的详细情况，我们将通过专业的研究、有关数据的分析等多种方式，同时也不会遗漏关于Anaconda Numpy 错误“Importing the Numpy C Extension Failed”是否有另一种解决方案、cvxpy 和 numpy 之间的版本冲突：“针对 API 版本 0xe 编译的模块，但此版本的 numpy 是 0xd”、Dask Distributed：对未出现在任务图可视化工具 (Bokeh) 中的 Dask 对象的操作、Dask:read_sql_table 为 KeyError + Dask 连接提供复合索引可用性的知识。

• 使用 Dask 进行内存 Numpy 计算的工作线程与线程
• Anaconda Numpy 错误“Importing the Numpy C Extension Failed”是否有另一种解决方案
• cvxpy 和 numpy 之间的版本冲突：“针对 API 版本 0xe 编译的模块，但此版本的 numpy 是 0xd”
• Dask Distributed：对未出现在任务图可视化工具 (Bokeh) 中的 Dask 对象的操作
• Dask:read_sql_table 为 KeyError + Dask 连接提供复合索引可用性

使用 Dask 进行内存 Numpy 计算的工作线程与线程

如何解决使用 Dask 进行内存 Numpy 计算的工作线程与线程

以下脚本用于基准测试：

from dask.distributed import Client
import dask.array as da
import numpy as np
import numba as nb
import time
import sys
N = 20000
M = 20000
def f_np():
    return np.random.random((N,M)).mean()
@nb.jit
def f_nb():
    return np.random.random((N,M)).mean()
def f_da():
    return da.random.random((N,M),chunks=(5000,5000)).mean().compute()
def loop(l,n,f):
    ts = []
    for i in range(n):
        t0 = time.time()
        f()
        t1 = time.time()
        ts.append(t1-t0)
    print(f"{l}: {round(sum(ts)/n,3)}: {'',''.join(str(round(t,3)) for t in ts)}")
if __name__ == "__main__":
    t,w,m = tuple(int(a) for a in sys.argv[1:])
    client = Client(
        processes=True,threads_per_worker=t,n_workers=w,memory_limit=f''{m}GB''
    )
    loop("np",3,f_np)
    loop("nb",f_nb)
    loop("da",f_da)

性能对比：

# python test_np.py [Threads per Worker] [Workers] [Memory Limit]
# what: AVG: run1,run2,run3
$ python test_np.py 1 1 12
np: 4.546: 4.521,4.474,4.644
nb: 4.125: 4.296,3.948,4.132
da: 4.698: 4.911,4.611,4.573
$ python test_np.py 4 1 12
np: 4.426: 4.408,4.448,4.422
nb: 3.938: 4.205,3.808,3.802
da: 2.064: 2.032,2.104,2.055
$ python test_np.py 1 4 3
np: 4.496: 4.498,4.533,4.456
nb: 3.996: 4.274,3.825,3.89
da: 2.076: 2.109,2.056,2.064
$ python test_np.py 4 4 3
np: 4.577: 4.612,4.565,4.556
nb: 4.029: 4.368,3.851,3.869
da: 2.077: 2.097,2.067,2.068

我试图了解这些情况在工作人员和线程数量方面的区别。情况 1 符合预期。

我的理解是，worker 的数量决定了并行计算的数量，线程数（每个 worker）决定了读/写操作的有效利用（在其他空闲时间期间）。

这就是为什么我是 f.x.想知道为什么案例 2 与案例 3 和案例 4 一样快。或者换句话说 - 为什么案例 3 和案例 4 没有案例 2 快，因为 IO 应该几乎可以忽略，因为一切都可以轻松地放入内存中。

关于观察 htop 案例 2 中的 cpu 活动与案例 3 和案例 4 无法区分。我希望案例 2 看起来与案例 1 相同 - 即只有一个活动核心。

计算是在 e2-standard-4（4 个 vcpu，16 GB 内存）GCE 实例上完成的。

$ uname -a
Linux dask 4.19.0-16-cloud-amd64 #1 SMP Debian 4.19.181-1 (2021-03-19) x86_64 GNU/Linux
$ python -V
Python 3.8.10
$ pip list | grep -E "dask|distributed|numpy|numba"
dask                2021.4.1
distributed         2021.4.1
numba               0.53.1
numpy               1.20.2

Anaconda Numpy 错误“Importing the Numpy C Extension Failed”是否有另一种解决方案

  \Traceback (most recent call last):
File "C:\Conda\envs\gcp\lib\site-packages\numpy\core\__init__.py",line 22,in <module>
from . import multiarray
File "C:\Conda\envs\gcp\lib\site-packages\numpy\core\multiarray.py",line 12,in <module>
from . import overrides
File "C:\Conda\envs\gcp\lib\site-packages\numpy\core\overrides.py",line 7,in <module>
from numpy.core._multiarray_umath import (
ImportError: DLL load Failed while importing _multiarray_umath: The specified module Could not be found.

During handling of the above exception,another exception occurred:

Traceback (most recent call last):
File "c:\API\citi-bike.py",line 4,in <module>
import numpy as np
File "C:\Conda\envs\gcp\lib\site-packages\numpy\__init__.py",line 150,in <module>
from . import core
File "C:\Conda\envs\gcp\lib\site-packages\numpy\core\__init__.py",line 48,in <module>
raise ImportError(msg)
ImportError:

IMPORTANT: PLEASE READ THIS FOR ADVICE ON HOW TO SOLVE THIS ISSUE!

Importing the numpy C-extensions Failed. This error can happen for
many reasons,often due to issues with your setup or how NumPy was
installed.

We have compiled some common reasons and troubleshooting tips at:

https://numpy.org/devdocs/user/troubleshooting-importerror.html

Please note and check the following:

* The Python version is: python3.9 from "C:\Conda\envs\gcp\python.exe"
* The NumPy version is: "1.21.1"

and make sure that they are the versions you expect.
Please carefully study the documentation linked above for further help.

Original error was: DLL load Failed while importing _multiarray_umath: The specified module Could not be found.

RuntimeError: module compiled against API version 0xe but this version of numpy is 0xd
Traceback (most recent call last):
  File "test.py",line 1,in <module>
    import cvxpy
  File "c:\Projekte\algo5\venv\lib\site-packages\cvxpy\__init__.py",line 18,in <module>
    from cvxpy.atoms import *
  File "c:\Projekte\algo5\venv\lib\site-packages\cvxpy\atoms\__init__.py",line 20,in <module>
    from cvxpy.atoms.geo_mean import geo_mean
  File "c:\Projekte\algo5\venv\lib\site-packages\cvxpy\atoms\geo_mean.py",in <module>
    from cvxpy.utilities.power_tools import (fracify,decompose,approx_error,lower_bound,File "c:\Projekte\algo5\venv\lib\site-packages\cvxpy\utilities\power_tools.py",in <module>
    from cvxpy.atoms.affine.reshape import reshape
  File "c:\Projekte\algo5\venv\lib\site-packages\cvxpy\atoms\affine\reshape.py",in <module>
    from cvxpy.atoms.affine.hstack import hstack
  File "c:\Projekte\algo5\venv\lib\site-packages\cvxpy\atoms\affine\hstack.py",in <module>
    from cvxpy.atoms.affine.affine_atom import AffAtom
  File "c:\Projekte\algo5\venv\lib\site-packages\cvxpy\atoms\affine\affine_atom.py",line 22,in <module>
    from cvxpy.cvxcore.python import canonInterface
  File "c:\Projekte\algo5\venv\lib\site-packages\cvxpy\cvxcore\python\__init__.py",line 3,in <module>
    import _cvxcore
ImportError: numpy.core.multiarray Failed to import

cvxpy 
numpy~=1.19.2 # tensorflow 2.4.1 requires this version

import cvxpy

if __name__ == ''__main__'':
    pass

# create dask array as example
x = da.random.random((10000,10000),chunks=(1000,1000))
y = x + x.T
z = y[::2,5000:].mean(axis=1)
z.compute()

import numpy as np
import pandas as pd
import sqlalchemy as sa
import dask.dataframe as ddf
start_time = datetime.Now() 
sqluri = f''oracle+cx_oracle://test:test@PROD1:1522/?service_name=prddb''
engine = sa.create_engine(sqluri)
daskDF = ddf.read_sql_table(table=''test1'',uri=sqluri,index_col=''ID'',schema=''testusr'')

sa_Meta = sa.MetaData()
sa_table = sa.Table("test1",sa_Meta,autoload=True,autoload_with=engine)
sa_query= sa.select("select /*+ ParaLLEL(8) */ a.*,a.id||a.period as d_idx from test1 a join test2 b on test1.id=test2.id where a.acct=''Client''")
ddf = dd.read_sql_table(sa_query,sqluri,index_col="d_idx ")