如何在Java应用里集成Spark MLlib训练好的模型做预测

前言

昨天媛媛说,你是不是很久没写博客了。我说上一篇1.26号,昨天3.26号,刚好两个月,心中也略微有些愧疚。今天正好有个好朋友问,怎么在Java应用里集成Spark MLlib训练好的模型。在StreamingPro里其实都有实际的使用例子,但是如果有一篇文章讲述下,我觉得应该能让更多人获得帮助

追本溯源

记得我之前吐槽过Spark MLlib的设计,也是因为一个朋友使用了spark MLlib的pipeline做训练,然后他把这个pipeline放到了spring boot里,结果做预测的时候奇慢无比,一条记录inference需要30多秒。为什么会这么慢呢?原因是Spark MLlib 是以批处理为核心设计理念的。比如上面朋友遇到的坑是有一部分原因来源于word2vec的transform方法:

@Since("2.0.0")
  override def transform(dataset: Dataset[_]): DataFrame = {
    transformSchema(dataset.schema, logging = true)
    val vectors = wordVectors.getVectors
      .mapValues(vv => Vectors.dense(vv.map(_.toDouble)))
      .map(identity) // mapValues doesn't return a serializable map (SI-7005)
    val bVectors = dataset.sparkSession.sparkContext.broadcast(vectors)
    val d = $(vectorSize)

来一条数据(通常API应用都是如此),他需要先获得vectors(词到vector的映射)对象,假设你有十万个词,

def getVectors: Map[String, Array[Float]] = {
    wordIndex.map { case (word, ind) =>
      (word, wordVectors.slice(vectorSize * ind, vectorSize * ind + vectorSize))
    }
  }

每次请求他都要做如上调用和计算。接着还需要把这些东西(这个可能就比较大了,几百M或者几个G都有可能)广播出去。

所以注定快不了。

把model集成到Java 服务里实例

假设你使用贝叶斯训练了一个模型,你需要保存下这个模型,保存的方式如下:

val nb = new NaiveBayes()
//做些参数配置和训练过程
.....
//保存模型
nb.write.overwrite().save(path + "/" + modelIndex)

接着,在你的Java/scala程序里,引入spark core,spark mllib等包。加载模型:

val model = NaiveBayesModel.load(tempPath)

这个时候因为要做预测,我们为了性能,不能直接调用model的transform方法,你仔细观察发现,我们需要通过反射调用两个方法,就能实现分类。第一个是predictRaw方法,该方法输入一个向量,输出也为一个向量。我们其实不需要向量,我们需要的是一个分类的id。predictRaw 方法在model里,但是没办法直接调用,因为是私有的:

  override protected def predictRaw(features: Vector): Vector = {
    $(modelType) match {
      case Multinomial =>
        multinomialCalculation(features)
      case Bernoulli =>
        bernoulliCalculation(features)
      case _ =>
        // This should never happen.
        throw new UnknownError(s"Invalid modelType: ${$(modelType)}.")
    }
  }

所以我们需要通过反射来完成:

val predictRaw = model.getClass.getMethod("predictRaw", classOf[Vector]).invoke(model, vec).asInstanceOf[Vector]

现在我们已经得到了predctRaw的结果,接着我们要用raw2probability 把向量转化为一个概率分布,因为spark 版本不同,该方法的签名也略有变化,所以可能要做下版本适配:

val raw2probabilityMethod = if (sparkSession.version.startsWith("2.3")) "raw2probabilityInPlace" else "raw2probability"
val raw2probability = model.getClass.getMethod(raw2probabilityMethod, classOf[Vector]).invoke(model, predictRaw).asInstanceOf[Vector]

raw2probability 其实也还是一个向量,这个向量的长度是分类的数目,每个位置的值是概率。所以所以我们只要拿到最大的那个概率值所在的位置就行:

val categoryId = raw2probability.argmax

这个时候categoryId 就是我们预测的分类了。

截止到目前我们已经完成了作为一个普通java/scala 方法的调用流程。如果我不想用在应用程序里,而是放到spark 流式计算里呢?或者批处理也行,那么这个时候你只需要封装一个UDF函数即可:

val models = sparkSession.sparkContext.broadcast(_model.asInstanceOf[ArrayBuffer[NaiveBayesModel]])
val f2 = (vec: Vector) => {
      models.value.map { model =>
        val predictRaw = model.getClass.getMethod("predictRaw", classOf[Vector]).invoke(model, vec).asInstanceOf[Vector]
        val raw2probability = model.getClass.getMethod(raw2probabilityMethod, classOf[Vector]).invoke(model, predictRaw).asInstanceOf[Vector]
        //model.getClass.getMethod("probability2prediction", classOf[Vector]).invoke(model, raw2probability).asInstanceOf[Vector]
        raw2probability
      }
    }

    sparkSession.udf.register(name , f2)

上面的例子可以参考StreamingPro 中streaming.dsl.mmlib.algs.SQLNaiveBayes的代码。不同的算法因为内部实现不同,我们使用起来也会略微有些区别。

总结

Spark MLlib学习了SKLearn里的transform和fit的概念,但是因为设计上还是遵循批处理的方式,实际部署后会有很大的性能瓶颈,不适合那种数据一条一条过来需要快速响应的预测流程,所以需要调用一些内部的API来完成最后的预测。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏算法修养

矩阵快速幂小结

      矩阵快速幂大概是用来解决这样一类问题,当你知道了一个递推式比如a[n]=a[n-1]+a[n-2] 题目要求你求出a[n]。如果n大于1亿怎么办? ...

3125
来自专栏机器学习算法与Python学习

机器学习(31)之频繁集挖掘FP Tree详解

关键字全网搜索最新排名 【机器学习算法】:排名第一 【机器学习】:排名第一 【Python】:排名第三 【算法】:排名第四 明早7:22推送第2期免费送书活动 ...

4196
来自专栏数据小魔方

IF函数——放松工作,享受生活!

今天跟大家分享一个简单却实用、高效的逻辑函数——IF函数。 ▼ IF函数可以简化很多我们数据处理过程中的重复性操作工作,让我们的工作效率大大提高。今天通过两个例...

3225
来自专栏杨建荣的学习笔记

圆周率π是怎么算出来的,用程序怎么算

下午在看一个算法的时候,突然看到了一个关于圆周率的问题,如果问你圆的周长怎么算,你肯定毫不犹豫是2πR,但是π是怎么算出来的呢?估计我们都没有想过,所以...

1773
来自专栏算法channel

玩转Pandas,让数据处理更easy系列6

玩转Pandas系列已经连续推送5篇,尽量贴近Pandas的本质原理,结合工作实践,按照使用Pandas的逻辑步骤,系统地并结合实例推送Pandas的主要常用功...

962
来自专栏数据小魔方

sparklines迷你图系列15——Composition(BoxPlot)

今天要跟大家分享的是sparklines迷你图系列14——BoxPlot。 箱线图是用于呈现数据分布形态(功能类似直方图)的一种图表,对于连续型数据,箱线图可以...

2964
来自专栏林德熙的博客

win2d 画出好看的图形

本文告诉大家,win2d 不需要从零开始做,以前做出来的很多库其实只需要做很小修改就可以做出好看的效果,而且用在 UWP 上。本文修改原先 大神写的 GDI 图...

1641
来自专栏潇涧技术专栏

Python Algorithms - C5 Traversal

Traversal就是遍历,主要是对图的遍历,也就是遍历图中的每个节点。对一个节点的遍历有两个阶段,首先是发现(discover),然后是访问(visit)。遍...

931
来自专栏aCloudDeveloper

算法导论第十九章 斐波那契堆

  《算法导论》第二版中在讨论斐波那契堆之前还讨论了二项堆,但是第三版中已经把这块的内容放到思考题中,究极原因我想大概是二项堆只是个引子,目的是为了引出斐波那契...

3238
来自专栏用户2442861的专栏

一之续、A*,Dijkstra,BFS算法性能比较及A*算法的应用

作者:July   二零一一年三月十日。 出处:http://blog.csdn.net/v_JULY_v ---------------------...

8691

扫码关注云+社区

领取腾讯云代金券