SparkML使用逻辑回归算法对新闻多分类

一、准备数据

自从引进DataFrame之后，spark在ml方面，开始使用DataFrame作为RDD的上层封装，以屏蔽RDD层次的复杂操作，对应用开发者提供简单的DataFrame，以减少开发量。本文中使用的是spark 2.4 版本为基础 ， 构建从数据预处理、特征转换、模型训练、数据测试到模型评估的一整套处理流程。另外，经过综合考虑，本文分词方法选用 HanLP 分词工具（文档丰富、算法公开、代码开源，并且经测试分词效果比较好），数据使用的是从新闻网站爬取的新闻分类数据，使用了4个分类的数据（文化、财经、军事和体育），每个分类使用了1000条数据，每行一条数据，有4个字段（分类、标题、日期和内容），使用“\u00EF”作分割符。

二、特征工程部分

2.1 数据清洗转换

数据预处理步骤主要进行数据清洗、转换操作。首先从文件加载数据到RDD，然后按照分隔符进行切分数据，因为分类字段爬下来时没有进行清洗，在这里我们需要将分类提取出来，并按照分类字段过滤掉未提取到分类或者分类不正确的脏数据，然后转换为DataFrame，接下来指定每个字段的字段名。

主要代码如下：

注意：这里必须要添加隐式转换 【 import spark.implicits._ 】，否则不能调用到SparkSQL的toDF方法！

 /**
  * 清洗步骤, 可根据具体数据结构和业务场景的不同进行重写. 注意: 输出必须要有标签字段"label"
  *
  * @param filePath 数据路径
  * @param spark    sparkSession
  * @return 清洗后的数据, 包含字段: "label", "title", "time", "content"
  */
def clean(filePath: String, spark: SparkSession): DataFrame = {
  import spark.implicits._
 
  val textRDD = spark.sparkContext.textFile(filePath)
 
  val textDF = textRDD.flatMap { line =>
    //分隔数据
    val fields = line.split("\u00EF")
 
    if (null != fields && fields.length > 3) {
      val categoryLine = fields(0)
      val categories = categoryLine.split("\\|")
      val category = categories.last
 
      var label = "其他"
      if (category.contains("文化")) label = "文化"
      else if (category.contains("财经")) label = "财经"
      else if (category.contains("军事")) label = "军事"
      else if (category.contains("体育")) label = "体育"
      else {}
 
      val (title, time, content) = (fields(1), fields(2), fields(3))
      if (!label.equals("其他")) {
        Some(label, title, time, content)
      } else {
        None
      }
    } else None
  }.toDF("label", "title", "time", "content")
  
  textDF
}

2.2 类别标签转换向量

因为在数据清洗之后取出来的标签是字符串类型，我们需要转换成Spark ML 逻辑回归算法能够识别的Double类型，在这里需要进行转换。

/**
    * 将数据中的中文标签转换为索引形式输出
    *
    * @param data 输入数据
    * @return 标签索引模型, 模型增加字段: "indexedLabel"
    */
  def indexrize(data: DataFrame): StringIndexerModel = {
    val labelIndexer = new StringIndexer()
      .setInputCol("label")
      .setOutputCol("indexedLabel")
      .fit(data)
 
    labelIndexer
  }

2.3 分词

（因为HanLP分词模型太大，未上传至项目中，需要的请从HanLP发布页下载，然后解压后将data目录整个放到SparkML\dictionaries\hanlp\目录下即可）

在经过数据预处理之后，我们已经将数据转换成需要的DataFrame格式，并且对数据进行了清洗。接下来我们需要进行分词的操作，将新闻内容切分成一个个词语的形式，以便后续进行停用词去除以及转换为特征向量

/**
    * 分词过程，包括"分词", "去除停用词"
    *
    * @param data   输入数据
    * @param params 配置参数
    * @return 预处理后的DataFrame, 增加字段: "tokens", "removed"
    */
  def segment(data: DataFrame, params: PreprocessParam): DataFrame = {
    val spark = data.sparkSession
 
    //=== 分词
    val segmenter = new Segmenter()
      .isDelEn(params.delEn) //是否去除英语单词
      .isDelNum(params.delNum) //是否去除数字
      .setSegmentType(params.segmentType) //分词方式
      .addNature(params.addNature) //是否添加词性
      .setMinTermLen(params.minTermLen) //最小词长度
      .setMinTermNum(params.minTermNum) //行最小词数
      .setInputCol("content")
      .setOutputCol("tokens")
 
    val segDF = segmenter.transform(data)
 
 
    //=== 去除停用词
    val stopwordArray = spark.sparkContext.textFile(params.stopwordFilePath).collect()
    val remover = new StopWordsRemover()
      .setStopWords(stopwordArray) //这是停用词集合
      .setInputCol(segmenter.getOutputCol)
      .setOutputCol("removed")
    val removedDF = remover.transform(segDF)
 
    removedDF
  }

这里我模仿spark ml包下的StopWordsRemover类创建了Segmenter类，用于对数据进行分词，其内部调用了HanLP分词工具。（由于spark自带的StopWordsRemover等使用的闭包仅限于ml包，自定义的类无法调用，故只是采用了与StopWordsRemover类似的使用形式，内部结构并不相同，并且由于以上原因，Segmenter类没有继承Transformer类，故无法进行pipeline管道操作，此缺陷有待解决）

Segmenter类具体实现如下：

package hnbian.sparkml.classnews.preprocess
 
import com.hankcs.hanlp.seg.Segment
import com.hankcs.hanlp.seg.common.Term
import com.hankcs.hanlp.tokenizer.{IndexTokenizer, NLPTokenizer, SpeedTokenizer, StandardTokenizer}
import hnbian.sparkml.classnews.utils.segment.{MyCRFSegment, MyNShortSegment}
import org.apache.spark.ml.util.Identifiable
import org.apache.spark.sql.{DataFrame, Row}
 
import scala.collection.JavaConversions._
 
 
/**
  * @author hnbian
  *         @ Description 基于hanLP的分词
  *         @ Date 2023/05/16 15:24
  **/
class Segmenter(val uid: String) extends Serializable {
  //英文字符正则
  private val enExpr = "[A-Za-z]+"
  //数值正则，可以匹配203,2.23,2/12
  private val numExpr = "\\d+(\\.\\d+)?(\\/\\d+)?"
  //匹配英文字母、数字、中文汉字之外的字符
  private val baseExpr =
    """[^\w-\s+\u4e00-\u9fa5]"""
 
  private var inputCol: String = ""
  private var outputCol: String = ""
  private var segmentType: String = "StandardTokenizer"
  private var addNature: Boolean = false
  private var delNum: Boolean = false
  private var delEn: Boolean = false
  private var minTermLen: Int = 1
  private var minTermNum: Int = 3
 
  def setInputCol(value: String): this.type = {
    this.inputCol = value
    this
  }
 
  def setOutputCol(value: String): this.type = {
    this.outputCol = value
    this
  }
 
  def setSegmentType(value: String): this.type = {
    this.segmentType = value
    this
  }
 
  def addNature(value: Boolean): this.type = {
    this.addNature = value
    this
  }
 
  def isDelNum(value: Boolean): this.type = {
    this.delNum = value
    this
  }
 
  def isDelEn(value: Boolean): this.type = {
    this.delEn = value
    this
  }
 
  def setMinTermLen(value: Int): this.type = {
    require(value > 0, "最小词长度必须大于0")
    this.minTermLen = value
    this
  }
 
  def setMinTermNum(value: Int): this.type = {
    require(value > 0, "行最小词数必须大于0")
    this.minTermNum = value
    this
  }
 
  def getInputCol: String = this.inputCol
 
  def getOutputCol: String = this.outputCol
 
  def getSegmentType: String = this.segmentType
 
  def this() = this(Identifiable.randomUID("segment"))
 
  def transform(dataset: DataFrame): DataFrame = {
    val spark = dataset.sparkSession
 
    var segment: Segment = null
    segmentType match {
      case "NShortSegment" =>
        segment = new MyNShortSegment()
      case "CRFSegment" =>
        segment = new MyCRFSegment()
      case _ =>
    }
 
    val tokens = dataset.select(inputCol).rdd.flatMap { case Row(line: String) =>
      var terms: Seq[Term] = Seq()
      segmentType match {
        case "StandardSegment" =>
          terms = StandardTokenizer.segment(line)
        case "NLPSegment" =>
          terms = NLPTokenizer.segment(line)
        case "IndexSegment" =>
          terms = IndexTokenizer.segment(line)
        case "SpeedSegment" =>
          terms = SpeedTokenizer.segment(line)
        case "NShortSegment" =>
          terms = segment.seg(line)
        case "CRFSegment" =>
          terms = segment.seg(line)
        case _ =>
          println("分词类型错误！")
          System.exit(1)
      }
 
 
      val termSeq = terms.flatMap { term =>
        val word = term.word
        val nature = term.nature
 
        if (this.delNum && word.matches(numExpr)) None //去除数字
        else if (this.delEn && word.matches(enExpr)) None //去除英文
        else if (word.length < minTermLen) None //去除过短的词
        else if (this.addNature) Some(word + "/" + nature)
        else Some(word)
      }
 
      if (termSeq.nonEmpty && termSeq.size >= minTermNum) Some((line, termSeq)) else None //去除词数过少的行
    }
 
    import spark.implicits._
    val tokensSet = tokens.toDF(inputCol + "#1", outputCol)
 
    dataset.join(tokensSet, dataset(inputCol) === tokensSet(inputCol + "#1")).drop(inputCol + "#1")
  }
}

主要在transform方法中调用了HanLP相关的分词方法。注意，如果使用NShortSegment和CRFSegment，需要new相应的对象，这里我自己创建了MyNShortSegment和MyCRFSegment类（代码如下），继承了HanLP中对应的类，并继承了Serializable特质（其实并没有做什么操作）。主要是因为HanLP没有对它们实现序列化，直接在RDD中使用它们会报错。（当然你也可以对HanLP的源码进行修改，再重新打包。个人觉得比较麻烦，并且不易跟进HanLP发布进度，所以没去弄）

package hnbian.sparkml.classnews.utils.segment
 
import com.hankcs.hanlp.seg.CRF.CRFSegment
 
 
/**
  * @author hnbian
  * @ Description 自定义CRF分词类，继承CRFSegment，并实现序列化
  * @ Date 2023/05/16 15:26
  **/
class MyCRFSegment extends CRFSegment with Serializable{
 
}

package hnbian.sparkml.classnews.utils.segment
 
import com.hankcs.hanlp.seg.NShort.NShortSegment
 
 
/**
  * @author hnbian
  * @ Description
  * @ Date 2023/05/16 15:27
  **/
class MyNShortSegment extends NShortSegment with Serializable {
 
}

2.4 停用词移除

分词之后，我们需要对一些常用的无意义词，如：“的”、“我们”、“是”等进行去除（通常去除的是语气词、连词等，统称为“停用词”）。因为这些词没有多大的意义，但是在自然语言中又经常使用，这些词不去掉会强烈的干扰我们对特征的抽取效果。（比如：在体育分类语料中，“的”共出现500次，“足球”共出现300次，那么谁更能代表体育这个分类呢？谁更应该作为特征被保留下来呢？）

去除停用词的操作我们直接调用ml包中的StopWordsRemover类：

//=== 去除停用词
    val stopwordArray = spark.sparkContext.textFile(params.stopwordFilePath).collect()
    val remover = new StopWordsRemover()
      .setStopWords(stopwordArray) //这是停用词集合
      .setInputCol(segmenter.getOutputCol)
      .setOutputCol("removed")
    val removedDF = remover.transform(segDF)

由于spark的StopWordsRemover类中内置的停用词都是一些英文停用词，而我们在这里处理的是中文语料，故需要加载自己的停用词。这里我使用了HanLPdictionary目录下的stopwords.txt文件提供的停用词。（这里面都是一些基本停用词，如果对停用词要求比较高，可以在网上找几份停用词表进行合并，效果会更好一点）

有兴趣的同学可以进到transform方法中看一看，spark官方的去除停用词方法跟我们常用处理一样，将停用词转换成set，然后调用contains进行判断，然后过滤：

@Since("2.0.0")
  override def transform(dataset: Dataset[_]): DataFrame = {
    val outputSchema = transformSchema(dataset.schema)
    val t = if ($(caseSensitive)) {
      val stopWordsSet = $(stopWords).toSet
      udf { terms: Seq[String] =>
        terms.filter(s => !stopWordsSet.contains(s))
      }
    } else {
      val lc = new Locale($(locale))
      val toLower = (s: String) => if (s != null) s.toLowerCase(lc) else s
      val lowerStopWords = $(stopWords).map(toLower(_)).toSet
      udf { terms: Seq[String] =>
        terms.filter(s => !lowerStopWords.contains(toLower(s)))
      }
    }
    val metadata = outputSchema($(outputCol)).metadata
    dataset.select(col("*"), t(col($(inputCol))).as($(outputCol), metadata))
  }

2.5 向量化

由于目前常用的分类、聚类等算法都是基于向量空间模型VSM（即将对象向量化为一个N维向量，映射成N维超空间中的一个点），VSM将数据转换为向量形式，便于对大规模数据进行矩阵操作等，也可以通过计算超空间中两个点之间的距离（一般是余弦距离）来计算两个向量之间的相似度。因此，我们需要将经过处理的语料转换为向量形式，这个过程叫做向量化。

这里我们也调用spark提供的向量化类CountVectorizer类进行向量化操作：

/**
    * 向量化过程, 包括词汇表过滤
    *
    * @param data   输入数据
    * @param params 配置参数
    * @return 向量模型
    */
  def vectorize(data: DataFrame, params: PreprocessParam): CountVectorizerModel = {
    //=== 向量化
    val vectorizer = new CountVectorizer()
      .setVocabSize(params.vocabSize) //特征词汇表大小
      .setInputCol("removed")
      .setOutputCol("features")
    val parentVecModel = vectorizer.fit(data)
 
    //过滤词汇表
    val numPattern = "[0-9]+".r
    val vocabulary = parentVecModel.vocabulary.flatMap { term =>
      if (term.length == 1 || term.matches(numPattern.regex)) None else Some(term)
    }
 
    val vecModel = new CountVectorizerModel(Identifiable.randomUID("cntVec"), vocabulary)
      .setInputCol("removed")
      .setOutputCol("features")
 
    //返回转换器模型
    vecModel
  }

这里的 vocabSize 是词汇表大小，即转换为向量之后的向量维度。通过阅读fit方法（训练向量化模型，主要是计算vocabulary词汇表的过程），我们可以看到其逻辑也比较简单：wordcount计算词频 –> 计算文档频率 –> 按文档频率过滤–>取词频最大的vocabSize个词

@Since("2.0.0")
  override def fit(dataset: Dataset[_]): CountVectorizerModel = {
    transformSchema(dataset.schema, logging = true)
    val vocSize = $(vocabSize)
    val input = dataset.select($(inputCol)).rdd.map(_.getAs[Seq[String]](0))
    val countingRequired = $(minDF) < 1.0 || $(maxDF) < 1.0
    val maybeInputSize = if (countingRequired) {
      Some(input.cache().count())
    } else {
      None
    }
    val minDf = if ($(minDF) >= 1.0) {
      $(minDF)
    } else {
      $(minDF) * maybeInputSize.get
    }
    val maxDf = if ($(maxDF) >= 1.0) {
      $(maxDF)
    } else {
      $(maxDF) * maybeInputSize.get
    }
    require(maxDf >= minDf, "maxDF must be >= minDF.")
    // 从这里开始 -----------------------------------
    val allWordCounts = input.flatMap { case (tokens) =>
      val wc = new OpenHashMap[String, Long]
      tokens.foreach { w =>
        wc.changeValue(w, 1L, _ + 1L) // 计算词频
      }
      wc.map { case (word, count) => (word, (count, 1)) }
    }.reduceByKey { case ((wc1, df1), (wc2, df2)) =>
      (wc1 + wc2, df1 + df2) //将每个词与在每篇文档中的词频相加，并将词的文档频率相加
    }
 
    val filteringRequired = isSet(minDF) || isSet(maxDF)
    val maybeFilteredWordCounts = if (filteringRequired) {
      //过滤出符合规则的词语（大于最小次数，小于等于最大次数）
      allWordCounts.filter { case (_, (_, df)) => df >= minDf && df <= maxDf }
    } else {
      allWordCounts
    }
 
    val wordCounts = maybeFilteredWordCounts
      .map { case (word, (count, _)) => (word, count) }
      .cache()
   // 到这里 -------------------------------------------------
    if (countingRequired) {
      input.unpersist()
    }
 
    val fullVocabSize = wordCounts.count()
 
   // 获取词频最多的 vocSize 个词，（不足的情况下按真实个数计算）然后按词频进行排序，保留排序后的词
    val vocab = wordCounts
      .top(math.min(fullVocabSize, vocSize).toInt)(Ordering.by(_._2))
      .map(_._1)
 
    require(vocab.length > 0, "The vocabulary size should be > 0. Lower minDF as necessary.")
    copyValues(new CountVectorizerModel(uid, vocab).setParent(this))
  }

从这里可以看出，所谓的训练CountVectorizer模型仅仅是对词频进行统计，计算出词频最大的 vocabSize 个词作为词汇表。下面我们继续看看transform方法：

@Since("2.0.0")
  override def transform(dataset: Dataset[_]): DataFrame = {
    transformSchema(dataset.schema, logging = true)
    // 将词汇表添加索引，转换为map，进行广播
    if (broadcastDict.isEmpty) {
      val dict = vocabulary.zipWithIndex.toMap
      broadcastDict = Some(dataset.sparkSession.sparkContext.broadcast(dict))
    }
    val dictBr = broadcastDict.get
    val minTf = $(minTF)
   // 向量稀疏化，将向量转为稀疏向量形式
    val vectorizer = udf { (document: Seq[String]) =>
      val termCounts = new OpenHashMap[Int, Double]
      var tokenCount = 0L
      document.foreach { term =>
        dictBr.value.get(term) match {
          // 遍历文档中的每个词，如果在该词词汇表中，则将其在词汇表 map 中的位置索引、词频总数做映射到termCounts 这个 map中
          case Some(index) => termCounts.changeValue(index, 1.0, _ + 1.0)
          case None => // ignore terms not in the vocabulary
        }
        tokenCount += 1
      }
      val effectiveMinTF = if (minTf >= 1.0) minTf else tokenCount * minTf
      // 按最小词频过滤，如果特征设置取二值特征，则将索引--> 词频 映射中的词频设置为1
      val effectiveCounts = if ($(binary)) {
        termCounts.filter(_._2 >= effectiveMinTF).map(p => (p._1, 1.0)).toSeq
      } else {
        termCounts.filter(_._2 >= effectiveMinTF).toSeq
      }
 
      Vectors.sparse(dictBr.value.size, effectiveCounts)
    }
    val attrs = vocabulary.map(_ => new NumericAttribute).asInstanceOf[Array[Attribute]]
    val metadata = new AttributeGroup($(outputCol), attrs).toMetadata()
    dataset.withColumn($(outputCol), vectorizer(col($(inputCol))), metadata)
  }

transform方法也比较简单，将词汇表建立索引并转换为Map –>遍历并保留在词汇表中的词，及其词频 –>转换为稀疏向量形式

我们可以将向量化后的数据打印出来看看长什么样儿：

/**
    * 向量化过程, 包括词汇表过滤
    *
    * @param data   输入数据
    * @param params 配置参数
    * @return 向量模型
    */
  def vectorize(data: DataFrame, params: PreprocessParam): CountVectorizerModel = {
    //=== 向量化
    val vectorizer = new CountVectorizer()
      .setVocabSize(params.vocabSize) //特征词汇表大小
      .setInputCol("removed")
      .setOutputCol("features")
    val parentVecModel = vectorizer.fit(data)
 
    //过滤词汇表
    val numPattern = "[0-9]+".r
    val vocabulary = parentVecModel.vocabulary.flatMap { term =>
      if (term.length == 1 || term.matches(numPattern.regex)) None else Some(term)
    }
 
    val vecModel = new CountVectorizerModel(Identifiable.randomUID("cntVec"), vocabulary)
      .setInputCol("removed")
      .setOutputCol("features")
 
    //返回转换器模型
    vecModel.transform(data).select("label","features").show(false)
    vecModel
  } 
 
+-----+------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|label|indexedLabel|features                                                                                                                                                                                                                                                                                                                                                                                                                    |
+-----+------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
|文化 |1.0         |(15000,[0,8,50,132,192,302,583,719,742,770,899,1274,2555,2677,2678,3396,4242,5172,5469,8100,8373,8546,9599,14688],[3.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,5.0,1.0,1.0,1.0,1.0,1.0])                                                                                                                                                                                                        |
|体育 |0.0         |(15000,[5,15,19,31,90,248,323,362,468,718,861,979,1098,1232,1233,1380,1507,1603,1852,2664,2958,3587,3689,4511,5354,5594,5775,7584,13147],[2.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,3.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,2.0,3.0,1.0,1.0,1.0,1.0])                                                                                                                                                             |
|体育 |0.0         |(15000,[7,14,15,56,71,129,174,191,222,258,368,385,419,446,447,523,643,703,752,786,923,967,1378,1588,3029,4159,5269,5642,5912,6645,8876],[1.0,1.0,1.0,1.0,1.0,1.0,1.0,5.0,1.0,2.0,1.0,1.0,1.0,5.0,3.0,1.0,1.0,1.0,1.0,1.0,6.0,2.0,1.0,3.0,1.0,1.0,4.0,1.0,1.0,1.0,2.0])                                                                                                                                                      |
|财经 |3.0         |(15000,[1,38,70,78,97,99,136,142,166,198,250,329,372,404,488,518,543,579,715,910,922,944,1045,1111,1162,1798,1825,1880,2559,3526,3848,4182,5462,7444,8564,8995,9222,10652,10703,10894,11646],[1.0,2.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,3.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0])                                                         |
|体育 |0.0         |(15000,[1,43,51,98,104,139,153,158,180,184,212,247,313,338,403,475,523,525,546,643,645,841,860,1126,1231,1249,1370,1696,1900,2325,3195,3382,3710,4111,4605,4975,5518,6932,7309,8587,8601,11573,12257,12616,12806],[1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,3.0,1.0,2.0,1.0,2.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,4.0,1.0,1.0])                    |
|财经 |3.0         |(15000,[1,19,33,46,59,114,126,131,137,175,233,247,271,287,324,332,474,523,550,698,704,789,794,828,836,1067,1190,1288,1565,1684,1889,1998,2338,3698,4028,4286,4837,5720,7011,7454,10655,11410,12263,12296,13151,14011,14703],[2.0,1.0,2.0,1.0,1.0,1.0,2.0,1.0,2.0,2.0,2.0,2.0,1.0,1.0,1.0,1.0,1.0,3.0,2.0,3.0,2.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,2.0,3.0,6.0,3.0,5.0,2.0,3.0,1.0,1.0,3.0,3.0,1.0,1.0,5.0,3.0,2.0,2.0])  |
+-----+------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

可以看到前面是中文标签，接着是我们经过转换的标签向量，即类别序号，后面是一个稀疏向量，其元素分别代表：向量维度(15000)、特征索引数组(即词汇表中哪些索引号的词出现在该文档中)、词频数组(词汇表中索引词在该文档中出现的次数)，例：第一行最后一个元素14688表示词汇表中第14688个词出现在该文档中，出现的次数是 1

三、模型训练与评估

3.1 模型训练

在经过向量化后，数据就可以用来进行分类模型的训练了！这里我们先使用最常用的分类模型——逻辑回归LogisticRegression。spark上提供的LR模型可以实现多分类，正好适用于我们的语料。

下面是分类模型训练的代码：

/**
  * LR模型训练处理过程, 包括: "模型训练"
  *
  * @param data   训练集
  * @return (向量模型, LR模型)
  */
  def train(data: DataFrame): LogisticRegressionModel = {
    val params = new ClassParam
 
    //=== LR分类模型训练
    data.persist()
    val lrModel = new LogisticRegression()
      .setMaxIter(params.maxIteration)  //模型最大迭代次数
      .setRegParam(params.regParam) //正则化项参数
      .setElasticNetParam(params.elasticNetParam) //L1范式比例, L1/(L1 + L2)
      .setTol(params.converTol) //模型收敛阈值
      .setLabelCol("indexedLabel")
      .setFeaturesCol("features")
      .fit(data)
    data.unpersist()
    this.saveModel(lrModel, params)
    lrModel.transform(data).select("label","prediction","probability").show(false)
    lrModel
  }
 
+------------+----------+---------------------------------------------------------------------------------+
|indexedLabel|prediction|probability                                                                      |
+------------+----------+---------------------------------------------------------------------------------+
|3.0         |3.0       |[0.16441041666862294,0.21278600323541705,0.19673717281009295,0.42606640728586703]|
|0.0         |0.0       |[0.6044368640741875,0.08082194053619135,0.06652880841860079,0.24821238697102038] |
|0.0         |3.0       |[0.24607915964277016,0.2656097748102712,0.20262095533491245,0.28569011021204616] |
|2.0         |2.0       |[0.10992902040055669,0.10823056085109142,0.6500273389120252,0.1318130798363267]  |
|1.0         |1.0       |[0.12108670218712453,0.6847084868394621,0.08512534822967509,0.10907946274373846] |
+------------+----------+---------------------------------------------------------------------------------+

在实例化一个LogisticRegression时，可以对其参数进行设置，这里大概跟大家说一下：

setMaxIter	设置最大迭代次数(默认100)，具体迭代过程可能会在不足最大迭代次数时停止(参照下一条)
setTol	设置容错(默认1E-6)，每次迭代会计算一个误差值，误差值会随着迭代次数的增加逐渐减小，如果误差值小于设置的容错值，则停止迭代优化
setRegParam	设置正则化项系数(默认0.0)，正则化项主要用于防止过拟合现象，因此，如果你的数据集比较小，特征维数又比较多时，易出现过拟合，此时可以考虑增大正则化项系数
setElasticNetParam	正则化范式比(默认0.0)，正则化一般有两种范式：L1(Lasso)和L2(Ridge)。L1一般用于特征的稀疏化，L2一般用于防止过拟合。这里的参数即设置L1范式的占比，默认0.0即只使用L2范式
setLabelCol	设置标签列(默认读取“label”列)
setFeaturesCol	设置特征列(默认读取“features”列)
setWeightCol	设置各个特征的权重（默认1.0），这里我们使用默认值就好了，如果对特征有特殊要求，可考虑重新设置对应的权重（如将标题作为一项特征，并且标题重要性更高，可将标题这一特征的权重设置大一点）

注意：由于我们的数据稀疏性本来就很高了(15000维的向量只有少数维度有值)，因此切记不要把setElasticNetParam设置得过大！！因为setElasticNetParam越大表示L1正则所占比例越高，对向量稀疏化效果越好，而我们的向量本来就很稀疏了，再稀疏化特征基本都为0了，得到的分类效果跟随机分类没什么区别(不信的话可以把这个值设置大一点，然后把后面说到的预测结果的probability打印出来，可以看到在各类别上的概率差别不大)

关于参数的设置，一般根据语料特点和业务场景的不同有所不同，这是一个经验性的东西，没有一个固定的计算公式（所以对数据挖掘和算法工程师来说，调参是一件相当耗时并且头疼的问题）。我们这里暂时使用spark官方example里面的设置，后面再进行调优。

23/05/16 11:50:04 INFO BlockManagerInfo: Removed broadcast_298_piece0 on DESKTOP-ROCPQ1G:7019 in memory (size: 28.6 KB, free: 799.7 MB)
23/05/16 11:50:04 INFO OWLQN: Step Size: 1.000
23/05/16 11:50:04 INFO OWLQN: Val and Grad Norm: 0.701753 (rel: 1.48e-13) 6.87906e-07
23/05/16 11:50:04 INFO OWLQN: Converged because gradient converged
23/05/16 11:50:04 INFO TorrentBroadcast: Destroying Broadcast(18) (from destroy at LogisticRegression.scala:817)
......

这里打印了每次迭代的步长(由算法内部自动设置)，以及每次迭代完后计算出的误差值，可以看到我们经过一定次数迭代后达到迭代次数上线，就停止迭代优化过程了。

刚才我们在代码中设置了打印前5个结果，可以看到console中有预测结果的输出：

+------------+----------+---------------------------------------------------------------------------------+
|indexedLabel|prediction|probability                                                                      |
+------------+----------+---------------------------------------------------------------------------------+
|3.0         |3.0       |[0.16441041666862294,0.21278600323541705,0.19673717281009295,0.42606640728586703]|
|0.0         |0.0       |[0.6044368640741875,0.08082194053619135,0.06652880841860079,0.24821238697102038] |
|0.0         |3.0       |[0.24607915964277016,0.2656097748102712,0.20262095533491245,0.28569011021204616] | // ---->分类错误
|2.0         |2.0       |[0.10992902040055669,0.10823056085109142,0.6500273389120252,0.1318130798363267]  |
|1.0         |1.0       |[0.12108670218712453,0.6847084868394621,0.08512534822967509,0.10907946274373846] |
+------------+----------+---------------------------------------------------------------------------------+

我们可以看到其中还是有一些分类预测错误的数据。

3.2 模型评估

虽说模型看上去效果不错，但是我们也需要一个量化指标来衡量其效果：这个模型的准确率、召回率和F1值(这3个指标是评判模型预测能力常用的一组指标，没听过的可以先去【模型评价指标：精确率、准确率、召回率、F1值】文章查看一下)。spark提供了用于多分类模型评估的类MulticlassClassificationEvaluator，我们就使用这个来测一测这个模型到底怎么样

具体代码如下：

package hnbian.sparkml.classnews.utils
 
import hnbian.sparkml.classnews.LRClassPredictDemo.predictions
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.sql.{DataFrame, Row}
 
/**
  * @author hnbian
  *         @ Description 多分类结果评估
  *         @ Date 2023/05/16 16:14
  **/
object Evaluations {
  /**
    * 评估结果
    *
    * @param data 分类结果
    * @return (准确率, 召回率, F1)
    */
  def multiClassEvaluate(data: DataFrame): (Double,Double, Double, Double) = {
    
    val evaluator = new MulticlassClassificationEvaluator()
      .setLabelCol("label")
      .setPredictionCol("prediction")
      .setMetricName("accuracy")
 
    val predictionRDD = predictions.select("prediction", "indexedLabel")
      .rdd.map { case Row(prediction: Double, label: Double) => (prediction, label) }
 
    val metrics = new MulticlassMetrics(predictionRDD)
 
    /** 准确率 */
    val accuracy = evaluator.evaluate(data)
    /** 精确率 */
    val weightedPrecision = metrics.weightedPrecision
    /** 召回率 */
    val weightedRecall = metrics.weightedRecall
    /** F1 */
    val f1 = metrics.weightedFMeasure
 
    (accuracy, weightedPrecision, weightedRecall, f1)
  }
}

运行结果如下：

========= 评估结果 ==========

准确率：0.9775172943889316
加权精确率：0.9777068320965895
加权召回率：0.9775172943889316
F1值：0.9775422468868562

看来效果确实还可以。准确率基本能达到97左右，这也是因为我们数据的类别区分度本身就比较好，如果选择的数据类别比较相近，分错的概率相对来说就比较大了。

整个测试的流程到这里基本结束了，一般数据挖掘的整体流程不外乎上面这些：

数据清洗转换 → 特征选择 → 向量化 → 模型选择与训练 → 模型测试 → 模型评估

四、调优

上面讲的是常规的处理流程，在使用算法的过程中，最初得到的结果往往并没有这么理想，这时我们需要对处理过程进行调优，接下来讲讲调优的事情。

4.1 调整训练集大小

训练集的大小将直接决定我们模型的好坏。一般情况下，用于模型训练的训练集应当越大越好（打个比方，如果让你猜一个东西是什么，是不是给的提示越多，越容易才出来？），如果训练集过小，极易导致过拟合，因此有时需要根据结果适当的将训练数据集调大，在我们的例子中对训练数据集的设置是10000，对于训练结果来说已经比较不错了。

过拟合：即模型在训练数据上准确率特别高，几乎都可以分对，但是对于新数据，其预测的准确率并不是很高，这时可以称这个模型的泛化能力差。导致过拟合的原因是数据量太少，训练时模型把个别数据的局部特征当成了全局特征来处理，比如说：如果我们就给模型几片带锯齿边缘的树叶，它可能得到的结果是树叶都带锯齿边缘，那么如果给他一片光滑边缘的树叶，模型可能就把它识别成不是树叶

4.2 特征选择

排除了过拟合，通过查看类别概率分布，发现每个文档在每个类别的概率相差不大，这意味着什么？可以猜测应该是我们的特征对样本数据的区分度不够，也就是说，使用目前选择的特征，无法很好地区分出哪些文档是属于哪个类别的！针对这种情况，我们先做以下两步操作：

过滤有效特征
增大特征维数
更改向量化方式
过滤有效特征
一般做特征选择时，都会尽量选择区分度大的特征，也就是容易从特征识别出是属于哪个类别（如从“足球”很容易看出应该是体育相关的文章）下面代码所示我们对向量做了哪些过滤与优化

//过滤词汇表过
    val numPattern = "[0-9]+".r
    val vocabulary = parentVecModel.vocabulary.flatMap { term =>
      //  1. 滤长度为1的词           2. 过滤数字
      if (term.length == 1 || term.matches(numPattern.regex)) None else Some(term)
    }

4.3 增加特征维数

特征区分效果不好，与特征数量也会有一定的影响，毕竟我们的语料是新闻长文本，每篇文档按200词计算，2000 * 4 *200 = 1600000，总共大于160万词，就算去除重复词、停用词，还应该有10W+的词量。这里我们将维度设置为 15000 对当前的数据预测基本上是恰当的，如果我们将维度设置过高预测准确度反而会下降。降低就是因为产生了过拟合，特征数量太多，模型的泛化能力下降，此时就可以确定准确率最高时的特征维数比较合适。这里由于篇幅问题，15000~50000中间的就省略就不继续测试了。（注意，模型训练的时间会随着特征维度的增加大幅增加，这是因为中间进行向量计算时，其计算量会因为向量长度增加成几何增长，这也就是我们常说的“维度灾难”）

4.4 改进向量化方式

在上面流程中，我们使用了根据词频来选择向量的特征，这是一种常用的方法，但是还有另一种更常用的方法——TF-IDF，中文叫做“文档-逆文档频率”，这里的文档频率其实就是我们上面用到的词频，逆文档频率其实就是预料中文档的总数除以包含该词的文档数，然后再取对数，具体公式如下：


这里词频除以出现次数最多的词的词频是为了做标准化处理，消除不同文档长短带来的影响（也可以除以当前文档词的总数），而求逆文档频率的时候将包含词的文档数+1是为了做平滑处理，防止出现除零的情况。

使用TF-IDF与直接使用词频做特征选择最大的不同是TF-IDF选出来的词的区分度更高，因为TF-IDF越高的词，代表这个词更加为当前文档所独有，因此更能代表这篇文档的属性。

由于我们这里的各项指标都已经很高了，将TF改成TF-IDF效果不是很大，故不做此步骤的优化！如果你的分类准确率并不是很高，可以替换成TF-IDF做特征选择，效果应该会有所提升，TF-IDF在spark中也已经提供，具体使用可参考这里（TF-IDF）（注意，该示例中使用了HashingTF来提取词频，但是该过程没有生成我们上述的词汇表，也就是说我们不能针对词频提取的特征进行过滤等操作，推荐把此部分更换为我们上述提到的使用CountVectorizer来做词频计算，然后再使用IDF方法提取IDF值）

4.5模型调参

我们这里使用到的LogisticRegression可以设置的参数在上面已经介绍过了，下面我们将针对这些参数进行调整，看看能否提高模型性能。

setMaxIter 与 setTol
这两个参数我们在上面也介绍过了，主要是用来控制模型迭代的次数。不知各位是否还记得，上面我们发现一个问题：使用40次迭代时，多次测试发现结果波动比较大，其实这个原因很明显：因为迭代次数不够，模型还没有收敛到最优，还处于波动状态，因此才会导致这个问题。如果我们设置迭代次数比较多，误差阈值比较小，这样虽然会延长模型训练的时间，但是训练处的模型会更加稳定，性能也会更优！

所以我们设置setMaxIter=100，setTol=1E-7

大家还可以对setRegParam和setElasticNetParam进行测试，这两个参数是控制正则化的，用于减小过拟合现象，这里我们就不进行测试了（如果数据本来就稀疏的情况下，增大setElasticNetParam可能会导致准确率下降！原因我们在上面参数说明的时候已经解释过了）