使用multiprocessing读取S3图片转数组出现RuntimeError该如何解决？

问题根源 #

RuntimeError: This class is not fork-safe 报错是因为你使用的S3文件操作客户端（比如s3fs、boto3封装的fs实例）属于非fork安全的类，Linux下Python multiprocessing默认使用fork模式启动子进程，会直接复制父进程的fs连接状态到子进程，导致连接异常崩溃。

解决方案 #

按照以下步骤修改即可正常运行：

• 1. 将多进程启动方式改为spawn，避免继承父进程的不安全连接
• 2. 不要使用全局fs实例，改为在子进程执行的函数内部初始化S3文件系统实例，每个子进程持有独立的S3连接
• 3. 修复原有代码的语法错误和逻辑错误

修正后代码 #

import multiprocessing
from PIL import Image
import s3fs
import pandas as pd
# 若使用TensorFlow的img_to_array则导入对应库，也可直接用PIL自带的toarray方法
from tensorflow.keras.preprocessing.image import img_to_array

# 全局配置仅放公共参数，不要初始化fs全局实例
bucket_name = "替换为你的S3桶名"
folder_to_images = "替换为你的S3图片路径前缀/"

def img_array(image):
    '''从S3读取单张图片并转为数组'''
    # 子进程内部初始化fs实例，每个子进程持有独立连接
    fs = s3fs.S3FileSystem()
    # 修复原with语句缺失的右括号
    with fs.open(f's3://{bucket_name}/{folder_to_images}{image}') as f:
        open_image = Image.open(f).resize((224,224))
        # 修复调用错误，原写法中image是文件名字符串，无法调用img_to_array方法
        img_arr = img_to_array(open_image)
    return img_arr

def add_to_df(n):
    '''返回两张图片对应的数组'''
    array_1 = img_array(n[0])
    # 修复逻辑错误，第二张图片取n[1]而非n[0]
    array_2 = img_array(n[1])
    return array_1, array_2

if __name__ == "__main__":
    # 启动方式改为spawn，必须放在main判断块内
    multiprocessing.set_start_method('spawn')
    # 此处替换为你自己的df加载逻辑
    df = pd.read_csv("你的数据文件路径")
    with multiprocessing.Pool() as p:
        result = p.map(add_to_df, df.values)
    # 将处理结果写入df对应列
    df[['array_1', 'array_2']] = pd.DataFrame(result, index=df.index)

注意事项 #

• 如果不需要依赖TensorFlow，也可以直接用PIL的open_image.resize((224,224)).toarray()实现数组转换，减少依赖
• spawn模式启动子进程的开销比fork略高，但可以完全避免非fork安全类的冲突问题，是处理S3、数据库连接这类资源的多进程标准方案

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