1、引言
最近，在做用户画像，利用文本分类方法挖掘用户兴趣模型。虽然文本分类不是很难，但是简单的事情，细节却是相当的重要。这篇文章我主要是想记录一下，我在做分类的时候，使用到的特征选择的方法，以及相关的是实现方法。
2、特征选择的方法
（1）信息增益
信息增益这一词来自通信领域，香浓提出的信息熵理论。信息熵的定义如下：
它的本质是衡量一个事件的不确定性的大小。而信息增益则是相对于某一个特定的特征而言的：例如，对与某个特征X，其对应的取值有n种（x1,x2,x3...xn），分别计算特征X在x1,x2,x3...xn的取值下的信息熵
，并根据每个取值的概率，计算出所有取值信息熵的平均值：（如下所示）
而信息增益就可以表示为原信息熵-条件熵。如下所示：
（2）信息增益率
特征取值的个数对信息增益有较大的干扰，为了避免有些特征取值个数较多，有些特征取值取值较小，造成信息增益无法公正的衡量一个特征的好坏。于是，信息增益率是在原特征的基础之上，相当与对信息增益做了归一化。其表达式是：
IGR=IG（T）/H(C)
（3）相关系数
相关系数是判断两个变量之间的相关性，比较常用的是person相关系数。
（4）Gini指数
基尼指数是表示一个变量的重要程度，其值越小越重要。
（5）卡方检验
卡方检验是检验两个变量之间是否相互独立的，在特征选择中的应用就是检验特征与目标变量之间的是否独立。在这里原价假设是特征与目标变量是相互独立的。统计计算特征的值与期望值之间的偏差，由于偏差不能取负值，所以计算特征的值与期望值之间的偏差的平方，并求和。最终结果作为判断原假设是否成立的标准。其公式如下：（这里需要注意一点就是期望值是我们自己计算得到的）
但是，在特征选择的过程中，我们需要选择的是原假设不成立的特征，即偏差较大的特征。以下介绍一下卡方检验在文本分类中的应用的例子：
假设现在有N篇文档，其中有M篇是关于体育的，我们想考察一个词“篮球”与类别“体育”之间的相关性。我们有四个观察值可以使用：
1. 包含“篮球”且属于“体育”类别的文档数，命名为A
2. 包含“篮球”但不属于“体育”类别的文档数，命名为B
3. 不包含“篮球”但却属于“体育”类别的文档数，命名为C
4. 既不包含“篮球”也不属于“体育”类别的文档数，命名为D
那么如何计算特征“篮球”的期望值呢？因为A+B是包含“篮球”的文章数，除以总文档数就是“篮球”出现的概率，当然，这里认为在一篇文章中出现即可，而不管出现了几次，而属于体育类的文章数为A+C，在这些个文档中，应该有
此时，我们已经得到了“篮球”这个特征的期望值，接下来就是计算实际值与期望值之间的偏差了。如下所示：
代入上是公式，可以得到：
但是，我们在实际工程上是通过其大小来选择特征词的，对于所有的特征词，A+C和B+D都是一样的，所以，上面的公式只需要计算如下式子即可：
最后我们选择其值较大的K个特征。
（6）通过建模的方式，来选择特征
该方法主要是通过模型的方式来选择特征，具体的做法：
1）首先根据原始数据，进行数据的预处理（如缺失值，异常值，归一化）
2）对原始数据进行建模，一般采用逻辑回归模型。
3）根据模型的评价指标（一般用正确率或AUC）来对模型进行预测调优。
4）若上述模型的评价指标较好，根据模型的权值的大小来确定特征的重要性。对应权值越大的特征，其重要性越大。
3、特征选择方法的实现
（1）信息增益率和相关系数：
1 package com.welab.BDL.UserInterests 2 3 import org.apache.spark.mllib.linalg.Vectors 4 import org.apache.spark.mllib.linalg.Vector 5 import org.apache.spark.SparkContext 6 import org.apache.spark.rdd.RDD 7 import org.apache.spark.SparkConf 8 import org.apache.spark.mllib.stat.Statistics 9 import org.apache.spark.mllib.linalg.Matrix 10 11 /** 12 * @author LJY 13 */ 14 object InformationGain { 15 16 //计算某个属性的信息熵 17 def inforEntropy(target_attribute: Array[Double]): Double = { 18 var temp = scala.collection.mutable.HashMap[Double, Int]() 19 for (item <- target_attribute) { 20 if (temp.contains(item)) { 21 temp(item) = temp(item) + 1 22 } else { 23 temp.put(item, 1) 24 } 25 } 26 var Entropy = 0.0 27 for (item <- temp) { 28 Entropy += (-1) * (item._2.toDouble / target_attribute.length) * log2(item._2.toDouble / target_attribute.length) 29 } 30 31 Entropy 32 } 33 34 def log2(x: Double): Double = scala.math.log(x) / scala.math.log(2) 35 36 //计算特征与目标特征之间的信息增益 37 def inforGain(sc: SparkContext, feature_attribute: RDD[(Double, Double)]): (Double, Double) = { 38 val target = feature_attribute.map { x => x._2 }.toArray() 39 val Entropy1 = inforEntropy(target) 40 41 val all_Entropy = sc.accumulator(0.0) 42 feature_attribute.groupBy(x => x._1).foreach { x => all_Entropy += (x._2.size.toDouble / target.length) * inforEntropy(x._2.map(x => x._2).toArray) 43 } 44 45 val X = feature_attribute.map { x => x._1 } 46 val Y = feature_attribute.map { x => x._2 } 47 48 val correlation: Double = Statistics.corr(X, Y, "pearson") 49 /* // calculate the correlation matrix using Pearson's method. Use "spearman" for Spearman's method. 50 // If a method is not specified, Pearson's method will be used by default. 51 val correlMatrix: Matrix = Statistics.corr(data, "pearson")*/ 52 53 // println(Entropy1) 54 // println(all_Entropy.value) 55 ((Entropy1 - all_Entropy.value), correlation) 56 } 57 58 //计算特征与目标特征之间的信息增益率 59 def inforGainRate(sc: SparkContext, feature_attribute: RDD[(Double, Double)]): (Double, Double) = { 60 val target = feature_attribute.map { x => x._2 }.toArray() 61 val Entropy1 = inforEntropy(target) 62 63 val all_Entropy = sc.accumulator(0.0) 64 feature_attribute.groupBy(x => x._1).foreach { x => all_Entropy += (x._2.size.toDouble / target.length) * inforEntropy(x._2.map(x => x._2).toArray) 65 } 66 67 val X = feature_attribute.map { x => x._1 } 68 val Y = feature_attribute.map { x => x._2 } 69 70 val correlation: Double = Statistics.corr(X, Y, "pearson") 71 /* // calculate the correlation matrix using Pearson's method. Use "spearman" for Spearman's method. 72 // If a method is not specified, Pearson's method will be used by default. 73 val correlMatrix: Matrix = Statistics.corr(data, "pearson")*/ 74 75 // println(Entropy1) 76 // println(all_Entropy.value) 77 ((Entropy1 - all_Entropy.value).toDouble/inforEntropy(X.toArray()), correlation) 78 } 79 80 def main(args: Array[String]): Unit = { 81 // Vectors.dense() 82 val conf = new SparkConf().setAppName("OneLevelClassification").setMaster("local") 83 val sc = new SparkContext(conf) 84 85 val data = sc.textFile("/user/hive/warehouse/bairong_summary") 86 87 val result = data.take(10).foreach { x => 88 val fields = x.split("\t") 89 fields.foreach { x => print(x + " ") } 90 } 91 92 } 93 }
（2）卡方检验自定义实现：
1 def my_chiSqTest3(sc: SparkContext, filename: String, p_thr: Double): Array[String] = { 2 val data = sc.textFile(filename, 10000).cache() 3 val allnum = data.count() 4 val result = data.map { x => 5 val fields = x.split("::") 6 val wds = fields(0).split(",") 7 val label = fields(1).split("#")(0) 8 var words = scala.collection.mutable.HashSet[String]() //准备对每个文档中的关键词进行去重 9 for (word <- wds) { 10 words.add(word) 11 } 12 var ss = "" 13 for (word <- words) { 14 if (ss.isEmpty()) { 15 ss = word + ":" + label 16 } else { 17 ss = ss + "," + word + ":" + label 18 } 19 } 20 ss 21 }.flatMap { x => x.split(",") }.map { x => 22 val fields = x.split(":") 23 (fields(0), fields(1)) //(word,label) 24 }.reduceByKey { (x, y) => x + "," + y }.cache() 25 26 println("所有的特征词的总数为：" + result.count()) 27 28 /* 29 * A ——表示包含某个关键词T，且属于某类X的文档数 30 * B ——表示包含某个关键词T，但不属于某类X的文档数 31 * C ——表示不包含某个关键词T，但属于某类X的文档数 32 * D ——表示即不包含某个关键词T，也不属于某类X的文档数 33 * A+C ——表示属于某类X的所有文档数 34 * C+D ——表示不包含关键词T的所有文档数 35 */ 36 37 val result_A_B = result.map { x => 38 val word = x._1 39 var label_occur = scala.collection.mutable.HashMap[String, Int]() //(label,occur) 40 val labels = x._2.split(",") //包含该关键词所有类别 41 var A = 0 42 var B = 0 43 //统计关键词在每个类别中出现的次数 44 for (label <- labels) { 45 label_occur.get(label) match { 46 case Some(a) => label_occur.put(label, a + 1) 47 case None => label_occur.put(label, 1) 48 } 49 } 50 //相对每个类来说，对应的A和B分别如下 51 var word_A_B = "" 52 for (label_num <- label_occur) { 53 A = label_num._2 54 B = labels.length - label_num._2 55 if (word_A_B.isEmpty()) { 56 word_A_B = word + ":" + label_num._1 + "," + A + "#" + B 57 } else { 58 word_A_B = word_A_B + "@" + word + ":" + label_num._1 + "," + A + "#" + B 59 } 60 61 } 62 word_A_B 63 }.flatMap { x => x.split("@") }.map { x => 64 val fields = x.split(",") 65 val word_label = fields(0) 66 val A = fields(1).split("#")(0).toInt 67 val B = fields(1).split("#")(1).toInt 68 (word_label, (A, B)) //(word:label,(A,B)) 69 } 70 71 val result_CplusD = result.map { x => 72 val word = x._1 73 val C_and_D = allnum - x._2.split(",").length 74 (word, C_and_D) //相对于关键词T来说，C+D 75 } 76 77 val result_Aplus_C = data.map { x => 78 val fields = x.split("::") 79 val label = fields(1).split("#")(0) 80 (label, 1) 81 }.reduceByKey(_ + _) 82 83 val res = result_A_B.map { x => 84 val word_label = x._1.split(":") 85 val word = word_label(0) 86 val label = word_label(1) 87 (word, (label, x._2)) //(word,(label,(A,B))) 88 }.leftOuterJoin(result_CplusD).map { x => 89 val word = x._1 90 val label_A_B = x._2._1 91 var C_plus_D = 0 92 x._2._2 match { 93 case Some(a) => C_plus_D = a.toInt 94 case None => 95 } 96 (label_A_B._1, (word, label_A_B._2, C_plus_D)) //(label,(word,(A,B),C+D)) 97 }.leftOuterJoin(result_Aplus_C).map { x => 98 val label = x._1 99 val word_A_B_CplusD = x._2._1 //(word,(A,B),C+D) 100 var AplusC = 0 101 x._2._2 match { 102 case Some(a) => AplusC = a 103 case None => 104 } 105 val A = word_A_B_CplusD._2._1 106 val B = word_A_B_CplusD._2._2 107 val C = AplusC - A 108 val D = word_A_B_CplusD._3 - C 109 (word_A_B_CplusD._1, ((A * D - B * C) * (A * D - B * C)).toDouble / ((A + B) * word_A_B_CplusD._3)) 110 }.reduceByKey { (x, y) => 111 var maxvalue = 0.0 112 if (x > y) { 113 maxvalue = x 114 } else { 115 maxvalue = y 116 } 117 maxvalue 118 }.map(x => (x._2, x._1)).sortByKey(false).map { x => (x._2, x._1) } 119 120 val res1 = res.take((res.count() * p_thr).toInt).map(x => x._1) 121 res1 122 }