使用多处理并行化 Keras 模型预测答案

【问题标题】：Parallelizing Keras Model Predict Using Multiprocessing使用多处理并行化 Keras 模型预测
【发布时间】：2022-01-26 11:24:42
【问题描述】：

我有一个有 60 个 CPU 的系统。我打算在多张图像上并行化 Keras 模型的预测。我尝试了以下代码：

img_model1 = tensorflow.keras.models.load_model('my_model.h5')
img_model2 = tensorflow.keras.models.load_model('my_model.h5')
img_model3 = tensorflow.keras.models.load_model('my_model.h5')
models=[img_model1,img_model2,img_model3] # all the three are same models

我尝试使用索引来避免弱引用酸洗错误：

def _apply_df(args):
    df, model_index = args
    preds=prediction_generator(df['image_path'])
    return  models[model_index].predict(preds)

def apply_by_multiprocessing(df, workers):
    workers = workers
    pool = Pool(processes=workers)
    result = pool.map(_apply_df, [(d,i) for i,d in enumerate(np.array_split(df[['image_path']], workers))])
    pool.close()
    return pd.concat(list(result))
    

apply_by_multiprocessing(df=data, workers=3)

代码会一直运行而不会产生任何结果... 我想这个问题可以用 tf.Sessions() 解决，但我不确定如何......

【问题讨论】：

标签： pandas keras multiprocessing pool weak-references

【解决方案1】：

在_apply_df 函数中加载您的模型，因此它不会参与酸洗和发送到流程。

这是一个没有使用pandas 的简单代码示例，它在时尚mnist 数据上运行模型。我认为您可以根据自己的用例进行调整。

import tensorflow as tf
import numpy as np
from multiprocessing import Pool


def _apply_df(data):
    model = tf.keras.models.load_model("my_fashion_mnist_model.h5")
    return model.predict(data)


def apply_by_multiprocessing(data, workers):

    pool = Pool(processes=workers)
    result = pool.map(_apply_df, np.array_split(data, workers))
    pool.close()
    return list(result)


def main():
    fashion_mnist = tf.keras.datasets.fashion_mnist
    _, (test_images, test_labels) = fashion_mnist.load_data()

    test_images = test_images / 255.0
    results = apply_by_multiprocessing(test_images, workers=3)
    print(test_images.shape)           # (10000, 28, 28)
    print(len(results))                # 3
    print([x.shape for x in results])  # [(3334, 10), (3333, 10), (3333, 10)]


if __name__ == "__main__":
    main()

【讨论】：

非常感谢。我一开始也试过这个，我刚刚在问题中添加了一些东西。这一次，问题是模型一直在运行！请检查新的编辑。谢谢
您是否尝试过我回答中的确切代码？它一直在运行吗？如果您能提供一个最小的运行示例，那就太好了，因为我无法重现此问题，并且我的代码在完成其工作后完成运行。
是的。我复制粘贴了你的代码。我所做的更改是使用此模型github.com/umbertogriffo/Fashion-mnist-cnn-keras/blob/master/…，因为我无法访问您的模型。我将数据重新整形为相应的输入形状 .reshape(10000,1,28,28)。是的。它一直在运行。如果没有并行化，我自己的代码可以工作。
您链接的模型是 TF 1 模型，而我使用的是 TF 2 模型。使用这个：drive.google.com/file/d/1dSBlDwy2ZdUaNAyEnOtm97OCQxP42Avk/… 另外，确保将您的外部代码（例如，apply_by_multiprocessing(df=data, workers=3)）放在main 函数中。尤其需要并行代码才能工作。
谢谢。 'spawn' 是没有必要的。我使用您的代码解决了这个问题，但此外，我必须从 .py 文件中导入池函数中使用的函数。事实证明，这个库的 .ipynb 有问题。再次感谢！