在 Juypter 中使用 cython 进行 Dask：ModuleNotFoundError：没有名为“_cython_magic”的模块

Question

我正进入（状态：

KilledWorker: ("('from_pandas-1445321946b8a22fc0ada720fb002544', 4)", 'tcp://127.0.0.1:45940')

我已经阅读了关于后一个错误消息的解释，但这与堆栈跟踪顶部的错误消息一起出现令人困惑：

distributed.utils - 错误 - Worker 已经存在 tcp://127.0.0.1:35780

实际错误通过管道传输到Jupyter notebook为我的笔记本运行命令的终端：

ModuleNotFoundError：没有名为“_cython_magic_faba6120a194ab58ae9efd1da474433f”的模块

所以我将自己研究如何解决这个问题，现在我在我的案例中发现了详细的错误。关于这种特殊配置的精确提示会很好，但我想将所有 cython 代码提取到笔记本外部的 python 代码中更明智，而不是敲击了解 cython 魔术命令？

Answer 1

这是一个完整的玩具示例（在 JupyterLab 上使用 SLURM 集群进行测试）。该示例使用 Cython 编译了一个对两个整数求和的普通函数，但当然可以将相同的技术应用于复杂（且更有用）的代码。
这里的关键技巧是必须设置 Workers 来查找和导入 Cython 库。
这需要在每个 Worker 上导入pyximport、调用pyximport.install()、然后导入 Cython 生成的模块。这是使用register_worker_callback(). 注意 Cython 生成的模块放置在<IPYTHONCACHEDIR/cython目录中（IPYTHONCACHEDIR可以通过调用找到IPython.paths.get_ipython_cache_dir()）。该目录必须添加到 Python 查找模块的路径中，以便可以加载 Cython 生成的模块。
这个例子假设了 SLURM，但这只是为了我的方便。dask.distributed “网络”可以用任何其他方法设置（例如参见http://distributed.dask.org/en/latest/setup.html）。

from dask import delayed

%load_ext cython

# Create a toy Cython function and put it into a module named remoteCython
%%cython -n remoteCython
def cython_sum(int a, int b):
    return a+b

# Set up a distributed cluster (minimal, just for illustration)
# I use SLURM.
from dask_jobqueue import SLURMCluster
from distributed import Client

cluster = SLURMCluster(memory="1GB",
                       processes=1,
                       cores=1,
                       walltime="00:10:00")

cluster.start_workers(1)   # Start as many workers as needed.



client = Client(cluster)

def init_pyx(dask_worker):
    import pyximport
    pyximport.install()

    import sys
    sys.path.insert(0,'<IPYTHONCACHEDIR>/cython/')   # <<< replace <IPYTHONCACHEDIR> as appropriate

    import remoteCython

client.register_worker_callbacks(init_pyx)  # This runs init_pyx() on any Worker at init

import remoteCython

# ASIDE: you can find where the full path of Cython-generated library by
# looking at remoteCython.__file__

# The following creates a task and submits to the scheduler.
# The task computes the sum of 123 and 321 via the Cython function defined above
future = client.compute(delayed(remoteCython.cython_sum)(123,321)) 

# The task is executed on the remote worker

# We fetch the result from the remote worker
print(future.result())   # This prints 444

# We're done. Let's release the SLURM jobs.
cluster.close()