如何使用在apachespark上训练的ML模型开发restapi？

Question

假设这种情况：

我们分析数据，使用手头的任何工具训练一些机器学习模型，并保存这些模型。这是在Python中完成的，使用apachespark pythonshell和API。我们知道apachespark擅长批处理，因此aboce场景是一个不错的选择。在

现在进入生产阶段，对于每个给定的请求，我们需要返回一个响应，该响应也取决于训练模型的输出。我认为这就是人们所说的流处理，而apacheflink通常被推荐使用。但是，如何在Flink管道中使用Python中可用的工具来训练相同的模型呢？在

Spark的微批处理模式在这里不起作用，因为我们确实需要响应每个请求，而不是成批响应。在

我还看到一些库试图在Flink中进行机器学习，但这不能满足那些在Python中使用不同工具而不是Scala，甚至不熟悉Scala的人的需求。在

所以问题是，人们如何处理这个问题？在

This问题是相关的，但不是重复的，因为作者在那里明确提到要使用Spark MLlib。这个库运行在JVM上，更有可能被移植到其他基于JVM的平台上。但是这里的问题是，如果人们使用scikit-learn，或者{}或者其他任何他们使用的方法/包，他们将如何处理它。在

Answer 1

看看MLeap。在

我们在将Spark学习到的模型外部化为单独的服务方面取得了一些成功，这些服务提供了对新传入数据的预测。我们外部化了LDA主题建模管道，尽管是在Scala中。但它们确实支持python，因此值得一看。在

Answer 2

我需要一种为ml Pipeline创建自定义Transformer的方法，并将该自定义对象与管道的其余部分一起保存/加载。这导致我深入研究spark模型序列化/反序列化的非常丑陋的深度。简而言之，所有的sparkml模型都有两个组件metadata和{}，其中模型数据是在.fit()期间学习到的参数。元数据保存在模型save dir下的一个名为metadata的目录中，据我所知是json，所以这不应该是个问题。模型参数本身似乎保存为save dir中的parquet文件。这是保存LDA模型的实现

override protected def saveImpl(path: String): Unit = {
      DefaultParamsWriter.saveMetadata(instance, path, sc)
      val oldModel = instance.oldLocalModel
      val data = Data(instance.vocabSize, oldModel.topicsMatrix, oldModel.docConcentration,
        oldModel.topicConcentration, oldModel.gammaShape)
      val dataPath = new Path(path, "data").toString
      sqlContext.createDataFrame(Seq(data)).repartition(1).write.parquet(dataPath)
    }

注意最后一行的sqlContext.createDataFrame(Seq(data)).repartition(1).write.parquet(dataPath)。好消息是，您可以将文件加载到您的web服务器中，如果服务器使用Java/Scala，您只需将spark jar保存在类路径中。在

但是，如果您在web服务器上使用python，那么您可以为python使用parquet库（https://github.com/jcrobak/parquet-python），坏消息是parquet文件中的一些或所有对象都将是二进制Java转储，因此您无法在python中实际读取它们。想到了一些选项，使用Jython（meh），使用Py4J加载对象，这就是pyspark用来与JVM通信的方式，这样就可以实际工作了。不过，我不认为这是一个非常简单的问题。在

或者从链接的问题中使用jpmml-spark并希望得到最好的结果。在