如何并行化Jupyter Notebook多输入批量处理？

下面针对你的需求，提供三种实用的进程级并行方案，均无需手动频繁转换Notebook为模块，同时解决全局变量依赖问题：

方案1：import_ipynb直接导入Notebook + multiprocessing并行 #

无需转换文件，直接将Notebook作为Python模块导入，配合进程池实现并行。

1. 安装依赖：

   pip install import_ipynb
   

2. 改造研究Notebook：
   将全局变量的定义封装成初始化函数，确保子进程能独立初始化变量：

   # 在你的研究Notebook（如research_notebook.ipynb）中添加
   def init_global_vars():
       global variable_01, variable_02, list_01, list_02, list_03
       variable_01 = ...  # 原常量定义
       variable_02 = ...
       list_01 = ...
       list_02 = fun_01()
       list_03 = fun_02(variable_01)
   
   # 主进程初始化一次，确保变量可用
   init_global_vars()
   

3. 新建调度Notebook实现并行：

   import import_ipynb
   import research_notebook as nb
   from multiprocessing import Pool
   
   def worker(input_pair):
       # 子进程独立初始化全局变量，避免进程间状态干扰
       nb.init_global_vars()
       input1, input2 = input_pair
       return nb.result_fun(input1, input2)
   
   # 准备多组输入
   input_pairs = [
       (val1_1, val2_1),
       (val1_2, val2_2),
       # 更多输入组
   ]
   
   # 启动进程池处理
   if __name__ == "__main__":
       with Pool(processes=4) as pool:  # processes设为CPU核心数
           results = pool.map(worker, input_pairs)
       # 处理输出结果
       print(results)
   

方案2：nbconvert自动转模块 + multiprocessing（适配Notebook更新） #

通过代码自动将Notebook转为Python模块，每次运行自动同步最新内容，避免手动转换。

1. 安装依赖：

   pip install nbconvert
   

2. 调度Notebook中添加自动转换逻辑：

   from nbconvert import PythonExporter
   import nbformat
   from multiprocessing import Pool
   
   def notebook_to_module(notebook_path, output_path):
       # 读取并转换Notebook为Python代码
       nb = nbformat.read(notebook_path, as_version=4)
       exporter = PythonExporter()
       source, _ = exporter.from_notebook_node(nb)
       with open(output_path, "w", encoding="utf-8") as f:
           f.write(source)
   
   # 自动转换研究Notebook为模块
   notebook_to_module("research_notebook.ipynb", "research_module.py")
   
   # 导入转换后的模块
   import research_module as rm
   
   def worker(input_pair):
       rm.init_global_vars()  # 同样需要初始化函数
       input1, input2 = input_pair
       return rm.result_fun(input1, input2)
   
   input_pairs = [...]  # 多组输入
   if __name__ == "__main__":
       with Pool(processes=4) as pool:
           results = pool.map(worker, input_pairs)
   

方案3：dask.distributed大规模并行（适合复杂计算） #

如果需要分布式扩展或更灵活的进程管理，dask提供进程级并行，支持进度可视化。

1. 安装依赖：

   pip install dask distributed
   

2. 调度Notebook实现并行：

   import import_ipynb
   import research_notebook as nb
   from dask.distributed import Client, delayed
   
   # 启动本地进程集群
   client = Client(n_workers=4)
   
   def worker(input_pair):
       nb.init_global_vars()
       input1, input2 = input_pair
       return nb.result_fun(input1, input2)
   
   input_pairs = [...]
   # 生成延迟任务
   tasks = [delayed(worker)(pair) for pair in input_pairs]
   # 执行任务并获取结果
   results = client.compute(tasks, sync=True)
   print(results)
   
   # 关闭集群
   client.close()
   

关键注意事项 #

• 全局变量处理：必须将全局变量初始化封装为函数，在每个子进程中调用。子进程内存空间独立，手动初始化能确保每个进程拥有正确的变量状态。
• 进程级并行：multiprocessing和dask均为进程级，不受GIL限制，完全适配CPU密集型任务。
• Notebook更新适配：方案1的import_ipynb会实时读取最新Notebook内容，方案2每次运行自动转换，均无需手动维护模块。

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