如何用Numba/Dask并行化Python调用bash脚本中的offline.exe循环？

首先明确：Numba不适合你的场景——它主要用于JIT编译Python数值计算代码，对调用外部offline.exe这类二进制程序的并行没有帮助，优先用Dask实现并行。

你的核心痛点是offline.exe依赖固定命名的输入文件，并行时会出现文件读写冲突，解决思路是为每个并行任务分配独立的工作目录，在目录内使用固定文件名执行模型，避免冲突。

具体实现步骤 #

1. 重构bash脚本run_offline.sh #

修改脚本，让它接受一个工作目录参数，进入该目录后执行模型循环，而非在当前目录运行：

#!/bin/bash
# 检查参数是否存在
if [ -z "$1" ]; then
    echo "Error: Please provide a working directory"
    exit 1
fi

WORK_DIR="$1"
# 进入工作目录并执行模型循环
cd "$WORK_DIR" || exit 1
for i in {1..9}; do
    ./offline.exe
done

2. 准备按时段命名的输入文件 #

将原循环中每个小时对应的OPTIONS.nam和PREP.nc，提前按时段命名为OPTIONS_YYYYMMDDHH.nam和PREP_YYYYMMDDHH.nc（比如2017年1月1日0点的文件为OPTIONS_2017010100.nam）。如果原代码是动态生成这两个文件，可将生成逻辑整合到后续Python任务函数中。

3. 用Dask实现并行处理 #

以下是Python代码示例，通过Dask的延迟任务和本地集群实现并行：

import os
import shutil
from tempfile import mkdtemp
from datetime import datetime, timedelta
from dask import delayed, compute
from dask.distributed import Client

# 生成2017-2019年所有小时的时间戳（格式：YYYYMMDDHH）
def generate_hourly_timestamps():
    timestamps = []
    current = datetime(2017, 1, 1, 0)
    end = datetime(2020, 1, 1, 0)
    while current < end:
        timestamps.append(current.strftime("%Y%m%d%H"))
        current += timedelta(hours=1)
    return timestamps

# 单个小时数据的处理任务
def process_hour(timestamp):
    # 创建临时工作目录
    work_dir = mkdtemp()
    try:
        # 复制对应时段的输入文件到临时目录，重命名为固定文件名
        src_opt = f"OPTIONS_{timestamp}.nam"
        src_prep = f"PREP_{timestamp}.nc"
        shutil.copy(src_opt, os.path.join(work_dir, "OPTIONS.nam"))
        shutil.copy(src_prep, os.path.join(work_dir, "PREP.nc"))
        
        # 调用修改后的bash脚本
        os.system(f"./run_offline.sh {work_dir}")
        
        # 复制输出文件到指定位置（假设模型输出为output.nc）
        output_dst = f"output_{timestamp}.nc"
        shutil.copy(os.path.join(work_dir, "output.nc"), output_dst)
    except Exception as e:
        print(f"Failed to process {timestamp}: {str(e)}")
    finally:
        # 清理临时目录
        shutil.rmtree(work_dir)
    return f"Processed {timestamp}"

if __name__ == "__main__":
    # 启动Dask本地集群，worker数量根据CPU核心数调整（比如8核设为8）
    client = Client(n_workers=4)
    
    timestamps = generate_hourly_timestamps()
    # 创建所有延迟任务
    tasks = [delayed(process_hour)(ts) for ts in timestamps]
    # 执行并行计算
    compute(tasks)
    
    client.close()

关键说明 #

• 临时目录的作用：每个并行任务拥有独立的输入文件副本，彻底避免多进程读写同一文件的冲突。
• Dask的优势：支持本地并行和集群扩展，自带任务调度、故障重试机制，适合处理大规模小时级任务。
• 参数调整：n_workers建议设置为机器CPU核心数，避免资源过度竞争；如果模型本身是多线程的，需适当减少worker数量。
• 错误处理：代码中添加了异常捕获，可记录失败的时段，方便后续重跑。

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