Python多进程并行写入报错求助：sed无法读取字符

首先，咱们来拆解你遇到的错误：sed: can't read r: No such file or directory这类报错，本质是sed收到的不是完整的文件路径，而是单个字符。这是因为你的多进程调用逻辑和run_sed函数的参数处理不匹配导致的。

核心错误点 #

你用p.map(run_sed, total_csv_file_list)的时候，multiprocessing.Pool.map会把total_csv_file_list里的**每个元素（单个CSV文件的完整路径）**依次传给run_sed函数。但你的run_sed函数里写了for filename in path:——这里的path是单个文件路径字符串，遍历字符串会把它拆成一个个字符（比如路径里的r、A、i、k），然后把这些字符当成文件名传给sed，自然会找不到文件！

修正后的代码 #

我帮你调整了代码逻辑，同时优化了路径处理和安全性：

import csv
import os
import multiprocessing as mp
from subprocess import run  # 比os.system更安全，能更好处理路径含空格的情况

path = '<somepath>'
# 提前构造完整的CSV文件列表，确保只筛选.csv文件（避免目录或其他格式文件）
total_csv_file_list = [
    os.path.join(path, 'csv_3', filename)
    for filename in os.listdir(os.path.join(path, 'csv_3'))
    if filename.endswith('.csv')
]
print(total_csv_file_list)

# 创建json目录（如果不存在），避免写入失败
json_dir = os.path.join(path, 'json')
os.makedirs(json_dir, exist_ok=True)

def process_single_csv(file_path):
    # 构造输出的JSON文件路径：放到指定的json目录，保持原文件名
    base_name = os.path.splitext(os.path.basename(file_path))[0]
    json_file_path = os.path.join(json_dir, f"{base_name}.json")
    
    # 用subprocess.run替代os.system，更安全且能捕获错误
    run([
        'sed',
        '1s/^/[/;$!s/$/,/;$s/$/]/',
        file_path,
        '-o', json_file_path
    ], check=True)

if __name__ == '__main__':  # Windows下多进程必须加这个判断，Linux/macOS也建议加
    with mp.Pool(processes=mp.cpu_count()) as p:
        p.map(process_single_csv, total_csv_file_list)

关键调整说明 #

• 修正函数参数逻辑：process_single_csv接收单个文件路径，不再遍历字符串，直接处理该文件。
• 安全的路径处理：用os.path模块统一处理路径，避免手动拼接字符串出错；提前创建json目录，确保写入权限。
• 替换os.system为subprocess.run：避免路径含空格时sed命令解析错误，同时check=True会在sed执行失败时抛出异常，方便调试。
• 添加if __name__ == '__main__'：这是Python多进程编程的最佳实践，尤其是Windows系统下必须加，防止子进程重复执行主模块代码。

额外提示 #

如果你的CSV文件内容里有特殊字符（比如引号），sed命令可能会生成格式错误的JSON。如果后续遇到JSON解析问题，可以考虑用Python原生的csv和json模块来处理，完全避免依赖shell命令，比如：

def process_single_csv_with_python(file_path):
    base_name = os.path.splitext(os.path.basename(file_path))[0]
    json_file_path = os.path.join(json_dir, f"{base_name}.json")
    
    with open(file_path, 'r', newline='', encoding='utf-8') as csv_f, \
         open(json_file_path, 'w', encoding='utf-8') as json_f:
        reader = csv.DictReader(csv_f)
        rows = list(reader)
        json.dump(rows, json_f, indent=2)

这种纯Python的方式兼容性更好，也更容易调试哦！

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