本文共 17688 字，大约阅读时间需要 58 分钟。
import org.apache.spark.mllib.classification.NaiveBayesimport org.apache.spark.mllib.evaluation.MulticlassMetricsimport org.apache.spark.mllib.feature.{HashingTF, IDF, Word2Vec}import org.apache.spark.{SparkConf, SparkContext}import org.apache.spark.mllib.linalg.{SparseVector => SV}import org.apache.spark.mllib.regression.LabeledPoint//使用别名引用/*** * 词频-逆文本频率(TF-IDF) * TF-IDF给一段文本（叫做文档）中每一个词赋予一个权值。这个权值是基于单词在文本中出现的频率（词频）计算得到的。 * 同时还要用逆向文本频率做全局的归一化。 * 逆向文本频率是指单词在所有文档（所有文档的集合对应的数据集通常称为文集）中的频率得到的。 * tf-idf(t,d) = tf(t,d) * idf(t) * 这里tf(t,d)是单词t在文档d中的频率（出现的次数）,idf(t)是文集中单词t的逆向文本频率， * idf(t)=log(N/d),N是文档的总数，d是出现过单词t的文档数量 * TF-IDF的含义是：在一个文档中出现次数很多的词相比出现次数少的词应该在词向量表得到更高的权值。 * 而IDF归一化起到了弱化在所有文档中总是出现的词的作用。最后的结果是，稀有的或者重要的词被赋予类更高的权重， * 而更加常用的单词则在考虑权重的时候有较小的影响。 *//*** * 特征哈希 * * 特征哈希是一种处理高维数据的技术，经常应用在文本和分类数据集上，这些数据集的特征可以取很多不同的值。 * 使用k分之一编码方式简单有效，但对于非常高维的数据时却不容易使用，因为它需要维护一个特征值到下标的映射； * 另外构建这个映射还要遍历一次数据集，这对于大规模数据集十分耗时，对并行场景扫描麻烦。 * 特征哈希是通过哈希方程对特征赋予向量下标，这个向量下标就是通过对特征的值做哈希得到的。 * 特征哈希的优势在于不再需要构建映射并把它保存在内存中。很容易实现、非常快。内存使用量不会随数据量和维度的增加而增加。 * *//*** * 20 Newsgroups由20个不同主题的新闻组消息组成的集合，有很多不同的数据格式。 * 这个数据集把可用数据拆分成训练集和测试集 */object CJNLP { def main(args: Array[String]): Unit = { //val sparkConf = new SparkConf().setAppName("SparkNLP").setMaster("local") //设置在本地模式运行 val BASEDIR = "hdfs://pc1:9000/20news-bydate-train/*" //HDFS文件 //val BASEDIR = "file:///home/chenjie/20news-bydate-train/*" // 本地文件 val sparkConf = new SparkConf() .setAppName("CJNLP-cluster") .setMaster("spark://pc1:7077") .setJars(List("untitled2.jar")) .set("spark.executor.memory", "3g") //设置在集群模式运行 val sc = new SparkContext(sparkConf) //初始化sc val rdd = sc.wholeTextFiles(BASEDIR)//wholeTextFiles加载的是文件和文本内容的打包 //加载数据 //下面分析20 newsgroup数据 val text = rdd.map{ case (file, text) => text}//wholeTextFiles加载的是文件和文本内容的打包 //text.cache() // 缓存到内存以提高速度 println(text.count) //11314 val newsgroups = rdd.map{ case (file, text) => file.split("/").takeRight(2).head //文件的全路径类似于/home/chenjie/20news-bydate-train/alt.atheism/49960 //将路径按/分开，并取右边的两个，再取右边两个中的第一个，即取倒数第二个，为该文件对应的新闻主题分类 } //去掉以下查看 val countByGroup = newsgroups.map{n => (n,1)}.reduceByKey(_ + _).collect().sortBy(- _._2).mkString("\n") //将新闻主题按主题统计个数并按从大到小排序 println(countByGroup) /*(rec.sport.hockey,600) (soc.religion.christian,599) (rec.motorcycles,598) (rec.sport.baseball,597) (sci.crypt,595) (sci.med,594) (rec.autos,594) (sci.space,593) (comp.windows.x,593) (sci.electronics,591) (comp.os.ms-windows.misc,591) (comp.sys.ibm.pc.hardware,590) (misc.forsale,585) (comp.graphics,584) (comp.sys.mac.hardware,578) (talk.politics.mideast,564) (talk.politics.guns,546) (alt.atheism,480) (talk.politics.misc,465) (talk.religion.misc,377)*/ //2、应用基本的分词方法 val whiteSpaceSplit = text.flatMap(t => t.split(" ").map(_.toLowerCase())) //应用基本的分词方法：空格分词，并把每个文档的所有单词变成小写 println(whiteSpaceSplit.distinct.count)//402978 //查看分词之后不同单词的数量 println(whiteSpaceSplit.sample(true, 0.3, 42).take(100).mkString(",")) //随机选择一篇文档看一下 /*die,the,academic,e,,in,prison,,if, >place,innocent,people,risk,than,if,state,error. i,higher,,,,,,,,,,do,that,type,and,i,kill,punishment,will,be,kill,kill,be,be,lacking,lacking,know",is,more,than,than,with,pretentions. ,case,university nntp-posting-host:,b64635.student.cwru.edu lines:,b64635.student.cwru.edu lines:,b64635.student.cwru.edu lines:,guy,said,just,rest,have,of,i,reproduce,is,a,big,file,file,i,misplaced,diskette,diskette,the,last,of,months,months,thanks,rather,is,,,angels,not,freewill., ,,,do., -- ,angels,not,,,do,what,them,,,mathew, t.split("""\W+""").map(_.toLowerCase)//把不是单词的字符过滤掉 ) println(nonWordSplit.distinct.count) //130126 println(nonWordSplit.distinct.sample(true, 0.3, 42).take(100).mkString(",")) //wuair,w1w3s1,42b,he3,hcq,6j,1pqd9hinnbmi,neurologists,believiing,jxicaijp,749,1472,eoeun,c1381,instantaneous,391k,typeset,typeset,bippy,hollombe,theoreticians,34ij,z0ozk,sunprops,sask,jesuit,6192,impute,1tbs,6jx,icbz,rlg1,9mf,cj1v,bowdoin,bowdoin,inre,inre,deadweight,deadweight,deterministic,createwindow,rockefeller,kjiv,kjiv,classifieds,ray_bourque,anachronistic,cherylm,005117,005117,005117,interfere,makewindow,mtearle,siiafeid8,moderates,x4_i,xtaddcallback,widmann,projector,jdecarlo,warms,triangulate,triangulate,recieves,eps,g45,g45,herod,1496,libpackagexcl,6w8rg,6w8rg,00ecgillespie,io21087,c4uzus,pdp11,ehs,placing,exxon,dxb132,hilly,8v0,023843,inconsitancies,isdres,trn,xa_rgb_default_map,fogbound,rchzd2_8d,mtagm,walters,r1865,lonny,lonny,arm6xx,likened,likened,rvik //随机选择一篇文档看一下 //发现效果好了一些，但仍有很多包含数字的单词剩下。下面过滤掉它们。 val regex = """[^0-9]*""".r val filterNumbers = nonWordSplit.filter(token => regex.pattern.matcher(token).matches )//使用正则模式过滤掉含有数字的单词 println(filterNumbers.distinct.count) //84912 //经此一役，单词集的大小再次减小 println(filterNumbers.distinct.sample(true, 0.3, 42).take(100).mkString(",")) //divisional,ntuvax,hem,gottschalk,semites,_congressional,hellenized,rlhzrlhz,tenex,ignore,_slightly_,mowtu,isgal,jbis,nondiscriminatory,steaminess,historians,historians,noport,cliche,bellevue,eur,claussen,vjpwu,dcbq,ja_jp,bippy,strut,brewmaster,searflame,hmih,nonnemacher,arresed,borg,ets,ets,subcircuits,subcircuits,sganet,sganet,nldp,internship,bombay,keysym,keysym,varda,connecters,handful,interconnecting,bhjn,bhjn,bhjn,worlders,dmitriev,butterfield,assemble,computational,sjoberg,kjiv,barbarity,silvers,antisemites,bombardments,emstation,emstation,jkis_ltd,twentieth,cfsmo,cfsmo,holdren,santiago,feszcm,rootx,rootx,springer,interfere,vow,formac,exhausting,fuenfzig,paradijs,systematically,hindenburg,diplomat,crudely,tossed,lastling,triangulate,intrepreted,plumbers,eps,ffbv,modifiable,tittle,tecmo,tecmo,overlapped,iauc,iauc,floor //可以看到所有数字字符已经被移除 //4、下面开始去除停用词，即and but the等 val tokenCounts = filterNumbers.map(t => (t,1)).reduceByKey(_ + _) //将所有单词计数 val oreringDesc = Ordering.by[(String, Int), Int](_._2) //按出现次数从大到小排序 println(tokenCounts.top(20)(oreringDesc).mkString("\n")) //可以看到 the to of ...等词出现最多，我们从中挑选出停用词表 /*(the,146532) (to,75064) (of,69034) (a,64195) (ax,62406) (and,57957) (i,53036) (in,49402) (is,43480) (that,39264) (it,33638) (for,28600) (you,26682) (from,22670) (s,22337) (edu,21321) (on,20493) (this,20121) (be,19285) (t,18728)*/ val stopwords = Set( "the","a","an","of","or","in","for","by","on","but","is","not", "with","as","was","if","they","are","this","and","it","have","from","at","my","be","that","to" )//停用词表 val tokenCountsFilteredStopWords = tokenCounts.filter{ case (k,v) => ! stopwords.contains(k) }//过滤掉停用词 println(tokenCountsFilteredStopWords.top(20)(oreringDesc).mkString("\n")) /*(ax,62406) (i,53036) (you,26682) (s,22337) (edu,21321) (t,18728) (m,12756) (subject,12264) (com,12133) (lines,11835) (can,11355) (organization,11233) (re,10534) (what,9861) (there,9689) (x,9332) (all,9310) (will,9279) (we,9227) (one,9008)*/ //观察到仍有停用词，这是因为我们停用词表不够大 val tokenCountsFilteredSize = tokenCountsFilteredStopWords.filter{ case (k,v) => k.size >= 2} //过滤掉仅仅含有一个字符的单词 println(tokenCountsFilteredSize.top(20)(oreringDesc).mkString("\n")) /* (ax,62406) (you,26682) (edu,21321) (subject,12264) (com,12133) (lines,11835) (can,11355) (organization,11233) (re,10534) (what,9861) (there,9689) (all,9310) (will,9279) (we,9227) (one,9008) (would,8905) (do,8674) (he,8441) (about,8336) (writes,7844) ... */ //5、下面基于频率去除单词：去掉在整个文本库中出现频率很低的单词 val oreringAsc = Ordering.by[(String, Int), Int](- _._2)//新建一个排序器，将单词，次数对按照次数从小到大排序 println(tokenCountsFilteredSize.top(20)(oreringAsc).mkString("\n")) /*(altina,1) (bluffing,1) (preload,1) (lennips,1) (actu,1) (vno,1) (wbp,1) (donnalyn,1) (ydag,1) (mirosoft,1) (jjjjrw,1) (harger,1) (conts,1) (bankruptcies,1) (uncompression,1) (d_nibby,1) (bunuel,1) (odf,1) (swith,1) (pacified,1)*/ //按出现频率从低到高排序 val rareTokens = tokenCounts.filter{ case (k, v) => v < 2}.map{ case (k, v) => k }.collect().toSet //将所有出现次数小于2的单词集合拿到 val tokenCountsFilteredAll = tokenCountsFilteredSize.filter{ case (k, v) => !rareTokens.contains(k) }//过滤掉所有出现次数小于2的单词 println(tokenCountsFilteredAll.top(20)(oreringAsc).mkString(",")) //(loyalists,2),(seetex,2),(upo,2),(jejones,2),(akl,2),(glorifying,2),(bxl,2),(singen,2),(sively,2),(petr_klima,2),(eoeun,2),(leymarie,2),(podsiadlik,2),(sloppiness,2),(kielbasa,2),(eer,2),(za_,2),(gottschalk,2),(pmu,2),(bisectors,2) println(tokenCountsFilteredAll.count)//打印不同的单词 //51801 //以上的过滤逻辑可以组合到一个函数中: def tokenize(line : String): Seq[String] = { line.split("""\W+""") .map(_.toLowerCase)//转为小写 .filter(token => regex.pattern.matcher(token).matches) .filterNot(token => stopwords.contains(token))//去掉停用词 .filterNot(token => rareTokens.contains(token))//去掉出现次数小于2的单词 .filter(token => token.size >= 2)//去掉字符数少于2的单词 .toSeq } println(text.flatMap(doc => tokenize(doc)).distinct().count()) //51801 //验证函数 val tokens = text.map(doc => tokenize(doc)) println(tokens.first().take(20)) //WrappedArray(mathew, mathew, mantis, co, uk, subject, alt, atheism, faq, atheist, resources, summary, books, addresses, music, anything, related, atheism, keywords, faq) //验证函数 //6、提取词干：可以把整个单词转换为一个基的形式。例如复数转为单数，过去时转为现在时等。这里先不予考虑 //7、训练TF-IDF模型 val dim = math.pow(2, 18).toInt val hashingTF = new HashingTF(dim) //HashingTF使用特征哈希把每个输入文本的词映射为一个词频向量的下标 //每个词频向量的下标是一个哈希值（依次映射到特征向量的某个维度）。词项的值是本身的TF-IDF权重 val tf = hashingTF.transform(tokens) //HashingTF的transform函数把每个输入文档（即词项的序列）映射到一个MLlib的Vector对象。 tf.cache() //把数据保持在内存中加速之后的操作 val v = tf.first().asInstanceOf[SV] println("tf的第一个向量大小:" + v.size) //262144 println("非0项个数：" + v.values.size) //706 println("前10列的下标：" + v.values.take(10).toSeq) //WrappedArray(1.0, 1.0, 1.0, 1.0, 2.0, 1.0, 1.0, 2.0, 1.0, 1.0) println("前10列的词频：" + v.indices.take(10).toSeq) //WrappedArray(313, 713, 871, 1202, 1203, 1209, 1795, 1862, 3115, 3166) //可以看到每一个词频的稀疏向量的维度是262144（2^18）.然而向量中的非0项只有706个。 val idf = new IDF().fit(tf) //创建IDF实例并fit，利用词频向量作为输入来对文库中的每个单词计算逆向文本频率 val tfidf = idf.transform(tf) //transform将词频向量转为TF_IDF向量 val v2 = tfidf.first().asInstanceOf[SV] println(v2.values.size) //706 println(v2.values.take(10).toSeq) //WrappedArray(2.3869085659322193, 4.670445463955571, 6.561295835827856, 4.597686109673142, 8.932700215224111, 5.750365619611528, 2.1871123786150006, 5.520408782213984, 3.4312512246662714, 1.7430324343790569) println(v2.indices.take(10).toSeq) //WrappedArray(313, 713, 871, 1202, 1203, 1209, 1795, 1862, 3115, 3166) //8、分析TF-IDF权重 //计算整个文档的TF-IDF最小和最大权值 val minMaxVals = tfidf.map{ v => val sv = v.asInstanceOf[SV] (sv.values.min, sv.values.max) } val globalMinMax = minMaxVals.reduce{ case ( (min1, max1), (min2, max2)) => (math.min(min1, min2), math.max(max1, max2)) } println(globalMinMax) //(0.0,66155.39470409753) //下面观察不同单词的IF-IDF权值。前面我们没有除移掉所有的停用词 val common = sc.parallelize(Seq(Seq("you","do","we"))) val tfCommon = hashingTF.transform(common) val tfidfCommon = idf.transform(tfCommon) val commonVector = tfidfCommon.first().asInstanceOf[SV] println(commonVector.values.toSeq) //WrappedArray(0.9965359935704624, 1.3348773448236835, 0.5457486182039175) //可以看到这些权重相对较低，也就算说tf-idf可以通过idf来过滤掉所有文档中都大量出现的词（停用词就是这样的词）的影响 val uncommon = sc.parallelize(Seq(Seq("telescope", "legislation", "investment"))) val tfUncommon = hashingTF.transform(uncommon) val tfidfUncommon = idf.transform(tfUncommon) val uncommonVector = tfidfUncommon.first().asInstanceOf[SV] println(uncommonVector.values.toSeq) //WrappedArray(5.3265513728351666, 5.308532867332488, 5.483736956357579) //9.3、使用TF—IDF模型 //预计两个从曲棍球新闻组选择的新闻比较相似，下面验证 val hockeyText = rdd.filter{ case (file, text) => file.contains("hockey") }//取出所有曲棍球新闻组 val hockeyTF = hockeyText.mapValues(doc => hashingTF.transform(tokenize(doc))) val hockeyTfIdf = idf.transform(hockeyTF.map(_._2)) import breeze.linalg._ val hockey1 = hockeyTfIdf.sample(true, 0.1, 42).first().asInstanceOf[SV] val breeze1 = new SparseVector(hockey1.indices, hockey1.values, hockey1.size) val hockey2 = hockeyTfIdf.sample(true, 0.1, 43).first().asInstanceOf[SV] val breeze2 = new SparseVector(hockey2.indices, hockey2.values, hockey2.size) val cosineSim = breeze1.dot(breeze2) / (norm(breeze1) * norm(breeze2)) println("两篇hockey文档余弦相似度为:" + cosineSim) //两篇hockey文档余弦相似度为:0.08332038223731995 //下面取一个别的新闻组中的文档来和曲棍球新闻组中的文档进行相似度计算 val graphicsText = rdd.filter { case (file, text) => file.contains("comp.graphics") } val graphicsTF = graphicsText.mapValues(doc => hashingTF.transform(tokenize(doc))) val graphicsTfIdf = idf.transform(graphicsTF.map(_._2)) val graphics = graphicsTfIdf.sample(true, 0.1, 42).first().asInstanceOf[SV] val breezeGraphics = new SparseVector(graphics.indices, graphics.values, graphics.size) val cosineSim2 = breeze1.dot(breezeGraphics) / (norm(breeze1) * norm(breezeGraphics)) println("一篇hockey文档和一篇graphics文档的相似度为：" + cosineSim2) //一篇hockey文档和一篇graphics文档的相似度为：0.011982956191037503 //相对于0.083这个值确实很低 //再取一篇baseball文档看看 val baseballText = rdd.filter { case (file, text) => file.contains("baseball") } val baseballTF = baseballText.mapValues(doc => hashingTF.transform(tokenize(doc))) val baseballTfIdf = idf.transform(baseballTF.map(_._2)) val baseball = baseballTfIdf.sample(true, 0.1, 42).first().asInstanceOf[SV] val breezeBaseball = new SparseVector(baseball.indices, baseball.values, baseball.size) val cosineSim3 = breeze1.dot(breezeBaseball) / (norm(breeze1) * norm(breezeBaseball)) println("cosineSim3=" + cosineSim3) //cosineSim3=0.013522460083035466 //3.2 基于20 Newsgroups数据集使用TF-IDF训练文本分类器 //当使用IF-IDF向量时，我们希望基于文档中共现的词语来计算余弦相似度，从而捕捉文档之间的相似度。 //类似地，我们也希望使用一个分类模型学习每个单词的权重，来得到某些单词出现（及出现）情况到特定主题的映射； //可以用来区分不同主题的文档 //也就是说学习到一个从某些单是否出现（和权重）到特定主题之间的映射关系 //在 20 Newsgroups的例子中，每一个新闻组的主题就算一个类，我们能使用TF-IDF转换后的向量作为输入训练一个分类器 //因为我们要处理的是一个多分类问题，我们使用MLlib中的朴素贝叶斯方法，这种方法支持多分类。 //选取20个主题并把它们转换到类的映射 val newsgroupMap = newsgroups.distinct().collect().zipWithIndex.toMap //从新闻组RDD开始，其中每个元素是一个话题，使用zipWithIndex,给每个类赋予一个数字下标 val zipped = newsgroups.zip(tfidf) //从新闻组RDD开始，其中每个元素是一个话题，使用zip函数把话题和由TF-IDF向量组成的tiidf RDD组合，其中每个label是一个类下标，特征就是IF-IDF向量 val train = zipped.map{ case (topic, vector) => LabeledPoint(newsgroupMap(topic), vector) } val model = NaiveBayes.train(train, lambda = 0.1) //使用朴素贝叶斯训练 val testPath = "hdfs://pc1:9000/20news-bydate-test/*" val testRDD = sc.wholeTextFiles(testPath) //加载测试集数据 val testLabes = testRDD.map{ case (file, text) => val topic = file.split("/").takeRight(2).head //将文件路径按/隔开，取最右边两个，再取其中的第一个，即全路径的倒数第二个，即为当前文件对于的类别 newsgroupMap(topic)//根据类别得到此类别的标号 } val testTf = testRDD.map{ case (file, text) => hashingTF.transform(tokenize(text)) //将每个文件的内容进行分词处理， //HashingTF的transform函数把每个输入文档（即词项的序列）映射到一个MLlib的Vector对象。 } val testTfIdf = idf.transform(testTf)//将MLlib的Vector对象转化为tf-idf向量 val zippedTest = testLabes.zip(testTfIdf)//将类型和向量打包 val test = zippedTest.map{ case (topic, vector) => LabeledPoint(topic, vector)//将类型和向量打包转换为LabeledPoint对象 } val predictionAndLabel = test.map(p => (model.predict(p.features), p.label)) //使用模型预测 val accuracy = 1.0 * predictionAndLabel.filter(x => x._1 == x._2).count() / test.count() //计算准确率 val metrics = new MulticlassMetrics(predictionAndLabels = predictionAndLabel) //计算加权F指标，是一个综合了准确率和召回率的指标（这里类似于ROC曲线下的面积，当接近1时有较好的表现），并通过类之间加权平均整合 println("准确率=" + accuracy) println("加权F指标=" + metrics.weightedFMeasure) //4、评估文本处理技术的作用 val rawTokens = rdd.map{ case(file, text) => text.split(" ")} val rawTF = rawTokens.map{ doc => hashingTF.transform(doc)} val rawTrain = newsgroups.zip(rawTF).map{ case (topic, vector) => LabeledPoint(newsgroupMap(topic), vector) } val rawModel = NaiveBayes.train(rawTrain, lambda = 0.1) val rawTestTF = testRDD.map{ case (file, text) => hashingTF.transform(text.split(" ")) } val rawZippedTest = testLabes.zip(rawTestTF) val rawTest = rawZippedTest.map{ case (topic, vector) => LabeledPoint(topic, vector) } val rawPredictionAndLabel = rawTest.map(p => (rawModel.predict(p.features), p.label)) val rawAccuracy = 1.0 * rawPredictionAndLabel.filter(x => x._1 == x._2).count() / rawTest.count() val rawMetrics = new MulticlassMetrics(rawPredictionAndLabel) println("原始模型准确率=" + rawAccuracy) println("原始模型F指标=" + rawMetrics.weightedFMeasure) //在以上的例子中，我们在用空格分词处理后的原始文本上应用哈希单词频率转换，并在这些文本上进行训练和评估 //发现尽管准确率和F指标比TF-IDF低几个百分点，但表现也还不错。 //5、Word2Vec模型 /*** * 另一类比较流行的模型是把每一个单词表示成一个向量。Word2Vec叫做分布向量表示。 * */ val word2vec = new Word2Vec() word2vec.setSeed(42) val word2VecModel = word2vec.fit(tokens) println("使用Word2Vec得到的hockey主题下的前20个词汇") word2VecModel.findSynonyms("hockey", 20).foreach(println) println("使用Word2Vec得到的legislation主题下的前20个词汇") word2VecModel.findSynonyms("legislation", 20).foreach(println) }} 转载地址：http://cuqrb.baihongyu.com/