在pyspark lambda映射函数中使用keras模型

我想使用该模型来预测 PySpark 中的映射 lambda 函数的分数。

def inference(user_embed, item_embed):
    feats = user_embed + item_embed
    dnn_model =  load_model("best_model.h5")
    infer = dnn_model.predict(np.array([feats]), verbose=0, steps=1)
    return infer
iu_score = iu.map(lambda x: Row(userid=x.userid, entryid=x.entryid, score = inference(x.user_embed, x.item_embed)))

运行速度极慢，代码运行后很快就卡在了最后阶段。

[Stage 119:==================================================>(4048 + 2) / 4050]

在 HTOP 监视器中，80 个核心中只有 2 个处于满负荷工作状态，其他核心似乎不工作。 那么我应该怎么做才能使模型并行预测呢？ iu 是 3 亿，所以效率对我来说很重要。 谢谢。

我已经转verbose=1，并且出现预测日志，但似乎预测只是一一预测，而不是并行预测。

在回答过程中我做了一些研究，发现这个问题很有趣。 首先，如果效率真的很重要，请花一点时间在没有 Keres 的情况下重新编码整个事情。您仍然可以使用张量流（模型）的高级 API，并且只需付出一点努力即可提取参数并将其分配给新模型。尽管包装器框架中的大量实现还不清楚（TensorFlow是一个不够丰富的框架吗？），但在升级时很可能会遇到向后兼容性的问题。确实不建议用于生产。

话虽如此，您能否检查一下到底是什么问题，例如，您是否使用 GPU？也许他们超载了？您能否将整个事情包装起来不超过某些容量并使用优先级系统？如果没有优先级，您可以使用简单的队列。您还可以检查是否确实终止了tensorflow的会话，或者同一台机器运行了许多干扰其他模型的模型。造成这种现象的原因有很多，如果能提供更多详细信息就太好了。

关于并行计算 - 您没有实现任何真正为该模型打开线程或进程的东西，所以我怀疑 pyspark 无法自行处理整个事情。也许实现（老实说我没有阅读整个 pyspark 文档）假设分派的函数运行得足够快并且没有按应有的方式分发。PySpark 只是映射缩减原理的复杂实现。分派的函数在单个步骤中扮演映射函数的角色，这对于您的情况可能会出现问题。尽管它是作为 lambda 表达式传递的，但您应该更仔细地检查哪些实例速度较慢，以及它们在哪些计算机上运行。

我强烈建议您执行以下操作：
Go to Tensorflow deplot 官方文档 https://www.tensorflow.org/deploy/并阅读如何真正部署模型。有一个用于与已部署模型进行通信的协议，称为RPC还有一个restful API。然后，使用 pyspark，您可以包装调用并与所提供的模型连接。你可以创建一个你想要的模型池，在 pyspark 中管理它，通过网络分配计算，从这里开始，天空和 cpus/gpus/tpus 是限制（我仍然对天空持怀疑态度）。

很高兴能从您那里得到有关结果的最新信息:)您让我很好奇。

我希望你能很好地解决这个问题，这是一个很好的问题。