Spark MLlib机器学习之朴素贝叶斯小试牛刀

太过爱你忘了你带给我的痛 2022-06-05 07:57 92阅读 0赞

### Spark MLlib机器学习之朴素贝叶斯小试牛刀 ###

废话不多说，直接上代码，以下代码均基于Scala实现。

##### **0. 机器学习工具类 BaseModelUtil.scala** #####

package com.jangz.recommendation.spark.mllib
    
    object BaseModelUtil { 
    
      private val DEFAULT_MODEL_PATH = "data/model"
    
      def modelPath(typ: String, path: String = DEFAULT_MODEL_PATH) = {
        path + "/" + typ
      }
    
      def main(args: Array[String]): Unit = {
        ChannelModel.main(args)
      }
    }

##### **1. 基于朴素贝叶斯的算法模型 ChannelModel.scala**  #####

package com.jangz.recommendation.spark.mllib
    
    import com.jangz.recommendation.util.SparkUtil
    import org.apache.hadoop.fs.Path
    import org.apache.hadoop.fs.FileSystem
    import scala.collection.mutable.ListBuffer
    import com.jangz.recommendation.util.FileUtil
    import com.jangz.recommendation.entity.Training
    import org.apache.spark.mllib.feature.HashingTF
    import org.apache.spark.mllib.regression.LabeledPoint
    import org.apache.spark.mllib.classification.NaiveBayes
    
    /** * Bias classification algorithm */
    object ChannelModel { 
    
      def main(args: Array[String]): Unit = {
        val sc = SparkUtil.getSparkContext(this.getClass)
    
        val modelPath = new Path(BaseModelUtil.modelPath("bayes"))
        val fs: FileSystem = FileSystem.get(modelPath.toUri, sc.hadoopConfiguration)
        if (fs.exists(modelPath)) {
          fs.delete(modelPath, true)
        }
    
        val list: ListBuffer[Training] = FileUtil.getTrainingList("data/ml/channel.txt")
        val arr = FileUtil.getTrainingArrayBuffer(list)
    
        val data = sc.parallelize(arr)
        val tf = new HashingTF(numFeatures = 10000)
        var parsedData = data.map(line => {
          var parts = line.split(',')
          LabeledPoint(parts(0).toDouble, tf.transform(parts(1).split(" ")))
        }).cache()
    
        val model = NaiveBayes.train(parsedData, lambda = 1.0, modelType = "multinomial")
    
        // save type
        model.save(sc, BaseModelUtil.modelPath("bayes"))
    
        sc.stop
      }
    }

##### **2. 朴素贝叶斯模型运用 ChannelAnalysis.scala**  #####

package com.jangz.recommendation.spark.mllib
    
    import com.jangz.recommendation.util.SparkUtil
    import com.jangz.recommendation.config.Config
    import org.apache.spark.mllib.classification.NaiveBayesModel
    import org.apache.spark.mllib.feature.HashingTF
    import com.jangz.recommendation.entity.Training
    import com.jangz.recommendation.util.FileUtil
    import com.jangz.recommendation.dao.TrainingDao
    
    object ChannelAnalysis { 
    
      def runAnalysis(): Unit = {
        val sc = SparkUtil.getSparkContext(this.getClass)
    
        val lines = sc.textFile(Config.input_path)
        val filter = SparkUtil.getFilterLog(lines).cache()
    
        val model = NaiveBayesModel.load(sc, BaseModelUtil.modelPath("bayes"))
        val tf = new HashingTF(numFeatures = 10000)
    
        /** * calculate channel classification infomation(map, reduce) */
        val map = filter.map(log => {
          val training: Training = new Training(pv = 1, uv = 1, ip = 1)
          training.channelId = model.predict(tf.transform(FileUtil.getTrainingString(log.getClearTitle()))).toInt
    
          training.uvs += log.Uuid
          training.ips += log.Ip
          (training.channelId, training)
        }).cache()
    
        /** * pv & uv & ip by general ReduceByKey */
        val reduce = map.reduceByKey((m, n) => {
          m.pv += n.pv
          m.uvs ++= n.uvs
          m.ips ++= n.ips
          m.uv = m.uvs.size
          m.ip = m.ips.size
          (m)
        })
    
        val list: List[Training] = reduce.values.collect().toList
        list.foreach(item => {
          item.day = Config.day
        })
        sc.stop()
    
        /** * save into DB */
        TrainingDao.saveChannelList(list)
      }
    
      def main(args: Array[String]): Unit = {
        val dayStr = if (args.length > 0) args(0) else "2016-12-01"
        Config.setDay(dayStr)
        runAnalysis()
      }
    }

该应用是用于新闻类型的分析。  
源代码可以前往[我的Github仓库下载][Github]

[Github]: https://github.com/Zychaowill/RecommendationSystem/tree/master/recommendation-analysis/src/main/scala/com/jangz/recommendation/spark/mllib