使用 Keras 模型作为 Apache Spark 和 Elephas 的广播变量答案

【问题标题】：Using Keras Model as a Broadcast variable with Apache Spark & Elephas使用 Keras 模型作为 Apache Spark 和 Elephas 的广播变量
【发布时间】：2018-03-11 14:32:54
【问题描述】：

我有一个预训练权重 [h5df] 约为 700mb 的 keras 模型。我想将它与 Apache Spark 一起用作广播变量。 1. 这似乎是不可能的，因为 keras 模型本身不具备火花感知能力且不可序列化。 2. 用谷歌搜索了一下，我发现 Elephas 库可以完成这项工作。所以尝试在 ElephasTransformer 中封装 Keras 预训练模型。这会引发多个错误（我使用 python 2.7）。例如在 Elephas 的 ml_model.py 文件中，“from pyspark.ml.util import keyword_only”，导入不可用。即使我尝试对此进行注释并在代码中进行适当的修改，它似乎也无法处理 keras 模型的广播，因为它在其中一个密集层中引发了张量错误。

如何在 apache spark 中使用预训练的 keras 模型作为广播变量？

【问题讨论】：

我成功共享了广播变量内的 900MB 数据。我想，对象成为广播变量的唯一要求是它应该是可序列化的。它可以在 Java 中轻松完成，对 python 不太确定。是否可以分享一些我们可以查看的代码？
你找到解决办法了吗？

标签： apache-spark keras broadcast

【解决方案1】：

我会使用SparkFiles 分发模型

spark.sparkContext.addFile("model_file.h5")

并在本地加载：

from pyspark import SparkFiles
from keras.models import load_model

def f(it):
    path = SparkFiles.get("mode_file.h5")
    model =  load.model(path)

    for i in it:
        yield ... # Do something


rdd.mapPartitions(f)

【讨论】：

您在实践中实施过这种方法吗？我很好奇它是否适合你。顺便说一句，您导入了load_model，但后来调用了load.model()。
这是唯一对我有用的方法。任何库都不支持带有自定义层的 keras 模型的序列化（到我已经搜索过的程度），所以我不得不遵循这个解决方案。

【解决方案2】：

在 Elephas 中，我解决此问题的方法是将权重设为广播变量，将 yaml 字符串作为参数提供给映射器函数，然后使用加载的 yaml 文件和权重在映射器函数内创建模型 -大意是：

from tensorflow.keras.models import model_from_yaml

weights = rdd.context.broadcast(model.get_weights())

def mapper_function(yaml_file, ...):
    model = model_from_yaml(yaml_file)
    model.set_weights(weights.value)

【讨论】：