Spark应用HanLP对中文语料进行文本挖掘--聚类详解教程

简介: 在Kmeans算法中,一个样本需要使用数值类型,所以需要把文本转为数值向量形式,这里在Spark中有两种方式。其一,是使用TF-IDF;其二,使用Word2Vec。这里暂时使用了TF-IDF算法来进行,这个算法需要提供一个numFeatures,这个值越大其效果也越好,但是相应的计算时间也越长,后面也可以通过实验验证。

软件:IDEA2014MavenHanLPJDK

用到的知识:HanLPSpark TF-IDFSpark kmeansSpark mapPartition;

用到的数据集:http://www.threedweb.cn/thread-1288-1-1.html(不需要下载,已经包含在工程里面);

工程下载:https://github.com/fansy1990/hanlp-test

1问题描述

现在有一个中文文本数据集,这个数据集已经对其中的文本做了分类,如下:

cebd16cee53a45614c735df82e59785b3ab8e80c

其中每个文件夹中含有个数不等的文件,比如环境有200个,艺术有248个;同时,每个文件的内容基本上就是一些新闻报道或者中文描述,如下:

8f717b0a59e6db4d92d7bd1e03f6390a9d864f07

现在需要做的就是,把这些文档进行聚类,看其和原始给定的类别的重合度有多少,这样也可以反过来验证我们聚类算法的正确度。

2.、解决思路:

2.1 文本预处理:

1. 由于文件的编码是GBK的,读取到Spark中全部是乱码,所以先使用Java把代码转为UTF8编码;  

2. 由于文本存在多个文件中(大概2k多),使用SparkwholeTextFile读取速度太慢,所以考虑把这些文件全部合并为一个文件,这时又结合1.的转变编码,所以在转变编码的时候就直接把所有的数据存入同一个文件中;

其存储的格式为: 每行: 文件名.txt\t文件内容

如:  41.txt   199601....

这样子的话,就可以通过.txt\t 来对每行文本进行分割,得到其文件名以及文件内容,这里每行其实就是一个文件了。

2.2 分词

分词直接采用HanLP的分词来做,HanLP这里选择两种:StandardNLP(还有一种就是HighSpeed,但是这个木有用户自定义词典,所以前期考虑先用两种),具体参考:https://github.com/hankcs/HanLP ;

2.3 词转换为词向量

Kmeans算法中,一个样本需要使用数值类型,所以需要把文本转为数值向量形式,这里在Spark中有两种方式。其一,是使用TF-IDF;其二,使用Word2Vec。这里暂时使用了TF-IDF算法来进行,这个算法需要提供一个numFeatures,这个值越大其效果也越好,但是相应的计算时间也越长,后面也可以通过实验验证。

2.4 使用每个文档的词向量进行聚类建模

在进行聚类建模的时候,需要提供一个初始的聚类个数,这里面设置为10,因为我们的数据是有10个分组的。但是在实际的情况下,一般这个值是需要通过实验来验证得到的。

2.5 对聚类后的结果进行评估

这里面采用的思路是:

1. 得到聚类模型后,对原始数据进行分类,得到原始文件名和预测的分类id的二元组(fileName,predictId)

2. 针对(fileNamepredictId),得到(fileNameFirstChar ,fileNameFirstChar.toInt - predictId)的值,这里需要注意的是fileNameFirstChar其实就是代表这个文件的原始所属类别了。

3. 这里有一个一般假设,就是使用kmeans模型预测得到的结果大多数是正确的,所以fileNameFirstChar.toInt-predictId得到的众数其实就是分类的正确的个数了(这里可能比较难以理解,后面会有个小李子来说明这个问题);

4. 得到每个实际类别的预测的正确率后就可以去平均预测率了。

5. 改变numFeatuers的值,看下是否numFeatures设置的比较大,其正确率也会比较大?

3、具体步骤:

3.1 开发环境--Maven

首先第一步,当然是开发环境了,因为用到了SparkHanLP,所以需要在pom.xml中加入这两个依赖:

1. <!-- 中文分词框架 -->

2.<dependency>

3.<groupId>com.hankcs</groupId>

4.<artifactId>hanlp</artifactId>

5.<version>${hanlp.version}</version>

6.</dependency>

7.<!-- Spark dependencies -->

8.<dependency>

9.<groupId>org.apache.spark</groupId>

10.<artifactId>spark-core_2.10</artifactId>

11.<version>${spark.version}</version>

12.</dependency>

13.<dependency>

14.<groupId>org.apache.spark</groupId>

15.<artifactId>spark-mllib_2.10</artifactId>

16.<version>${spark.version}</version>

17.</dependency>

其版本为:

<hanlp.version>portable-1.3.4</hanlp.version><spark.version>1.6.0-cdh5.7.3</spark.version>

3.2 文件转为UTF-8编码及存储到一个文件

 

这部分内容可以直接参考:src/main/java/demo02_transform_encoding.TransformEncodingToOne 这里的实现,因为是Java基本的操作,这里就不加以分析了。

3.3 Scala调用HanLP进行中文分词

Scala调用HanLP进行分词和Java的是一样的,同时,因为这里有些词语格式不正常,所以把这些特殊的词语添加到自定义词典中,其示例如下:

1.import com.hankcs.hanlp.dictionary.CustomDictionary

2.import com.hankcs.hanlp.dictionary.stopword.CoreStopWordDictionary

3.import com.hankcs.hanlp.tokenizer.StandardTokenizer

4.import scala.collection.JavaConversions._

5./**

6.* Scala 分词测试

7.* Created by fansy on 2017/8/25.

 8.*/

9.object SegmentDemo {

10.def main(args: Array[String]) {

11.val sentense = "41,【 日 期 】19960104 【 版 号 】1 【 标 题 】合巢芜高速公路巢芜段竣工 【 作 者 】彭建中 【 正 文 】 安徽合(肥)巢(湖)芜(湖)高速公路巢芜段日前竣工通车并投入营运。合巢芜 高速公路是国家规划的京福综合运输网的重要干线路段,是交通部确定1995年建成 的全国10条重点公路之一。该条高速公路正线长88公里。(彭建中)"

12.CustomDictionary.add("日 期")

13.CustomDictionary.add("版 号")

14.CustomDictionary.add("标 题")

15.CustomDictionary.add("作 者")

16.CustomDictionary.add("正 文")

17.val list = StandardTokenizer.segment(sentense)

18.CoreStopWordDictionary.apply(list)

19.println(list.map(x => x.word.replaceAll(" ","")).mkString(","))

20.}

21.}

运行完成后,即可得到分词的结果,如下:

e921c1178bb2fa769a96d6b53dfafa853603d98e

考虑到使用方便,这里把分词封装成一个函数:

1./**

2.* String 分词

3.* @param sentense

4.* @return

5.*/

6.def transform(sentense:String):List[String] ={

7.val list = StandardTokenizer.segment(sentense)

8.CoreStopWordDictionary.apply(list)

9.list.map(x => x.word.replaceAll(" ","")).toList

10.}

11.}

 

输入即是一个中文的文本,输出就是分词的结果,同时去掉了一些常用的停用词。

3.4 TF-IDF

Spark里面求TF-IDF,可以直接调用Spark内置的算法模块即可,同时在Spark的该算法模块中还对求得的结果进行了维度变换(可以理解为特征选择或“降维”,当然这里的降维可能是提升维度)。代码如下:

1.val docs = sc.textFile(input_data).map{x => val t = x.split(".txt\t");(t(0),transform(t(1)))}

2..toDF("fileName", "sentence_words")

3.

4.// 3. TF

5.println("calculating TF ...")

6.val hashingTF = new HashingTF()

7..setInputCol("sentence_words").setOutputCol("rawFeatures").setNumFeatures(numFeatures)

8.val featurizedData = hashingTF.transform(docs)

9.

10.// 4. IDF

11.println("calculating IDF ...")

12.val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")

13.val idfModel = idf.fit(featurizedData)

14.val rescaledData = idfModel.transform(featurizedData).cache()

变量docs是一个DataFrame[fileName, sentence_words] ,经过HashingTF后,变成了变量 featurizedData ,同样是一个DataFrame[fileName,sentence_words, rawFeatures]。这里通过setInputCol以及SetOutputCol可以设置输入以及输出列名(列名是针对DataFrame来说的,不知道的可以看下DataFrameAPI)。

接着,经过IDF模型,得到变量 rescaledData ,其DataFrame[fileName,sentence_words, rawFeatures, features]

执行结果为:

95436a395e45dc38e91424fab17ddf0e1b55a859

3.5 建立KMeans模型

 

直接参考官网给定例子即可:

1.println("creating kmeans model ...")

2.val kmeans = new KMeans().setK(k).setSeed(1L)

3.val model = kmeans.fit(rescaledData)

4.// Evaluate clustering by computing Within Set Sum of Squared Errors.

5.println("calculating wssse ...")

6.val WSSSE = model.computeCost(rescaledData)

7.println(s"Within Set Sum of Squared Errors = $WSSSE")

这里有计算cost值的,但是这个值评估不是很准确,比如我numFeature设置为2000的话,那么这个值就很大,但是其实其正确率会比较大的。

3.6 模型评估

这里的模型评估直接使用一个小李子来说明:比如,现在有这样的数据:

58cdaf44e11ec6d5c93241da8740219c8b800d4f

其中,1开头,2开头和4开头的属于同一类文档,后面的0,3,2,1等,代表这个文档被模型分类的结果,那么可以很容易的看出针对1开头的文档,

其分类正确的有4个,其中("123.txt",3)以及(“126.txt,1)是分类错误的结果,这是因为,在这个类别中预测的结果中0是最多的,所以0是和1开头的文档对应起来的,这也就是前面的假设。

1. 把同一类文档分到同一个partition中;

1.val data = sc.parallelize(t)

2.val file_index = data.map(_._1.charAt(0)).distinct.zipWithIndex().collect().toMap

3.println(file_index)

4.val partitionData = data.partitionBy(MyPartitioner(file_index))

这里的file_index,是对不同类的文档进行编号,这个编号就对应每个partition,看MyPartitioner的实现:

1.case class MyPartitioner(file_index:Map[Char,Long]) extends Partitioner

2.override def getPartition(key: Any): Int = key match {

3.case _ => file_index.getOrElse(key.toString.charAt(0),0L).toInt

4.}

 

5..override def numPartitions: Int = file_index.size

6.}

2. 针对每个partition进行整合操作:

在整合每个partition之前,我们先看下我们自定义的MyPartitioner是否在正常工作,可以打印下结果:

1.val tt = partitionData.mapPartitionsWithIndex((index: Int, it: Iterator[(String,Int)]) => it.toList.map(x => (index,x)).toIterator)

2.tt.collect().foreach(println(_))

运行如下:

c8ff83d11c2816722cd4f6334191e66e5ef81dcb

其中第一列代表每个partitionid,第二列是数据,发现其数据确实是按照预期进行处理的;接着可以针对每个partition进行数据整合:

1.// firstCharInFileName , firstCharInFileName - predictType

2.val combined = partitionData.map(x =>( (x._1.charAt(0), Integer.parseInt(x._1.charAt(0)+"") - x._2),1) )

3..mapPartitions{f => var aMap = Map[(Char,Int),Int]();

4.for(t <- f){

5.if (aMap.contains(t._1)){

6.aMap = aMap.updated(t._1,aMap.getOrElse(t._1,0)+1)

7.}else{

8.aMap = aMap + t

9.}

10.}

11.val aList = aMap.toList

12.val total= aList.map(_._2).sum

13.val total_right = aList.map(_._2).max

14.List((aList.head._1._1,total,total_right)).toIterator

15.// aMap.toIterator //打印各个partition的总结

16. }

在整合之前先执行一个map操作,把数据变成((fileNameFirstChar, fileNameFirstChar.toInt - predictId), 1),其中fileNameFirstChar代表文件的第一个字符,其实也就是文件的所属实际类别,后面的fileNameFirstChar.toInt-predictId 其实就是判断预测的结果是否对了,这个值的众数就是预测对的;最后一个值代码前面的这个键值对出现的次数,其实就是统计属于某个类别的实际文件个数以及预测对的文件个数,分别对应上面的totaltotal_right变量;输出结果为:

(4,6,3)

(1,6,4)

(2,6,4)

发现其打印的结果是正确的,第一列代表文件名开头,第二个代表属于这个文件的个数,第三列代表预测正确的个数

这里需要注意的是,这里因为文本的实际类别和文件名是一致的,所以才可以这样处理,如果实际数据的话,那么mapPartitions函数需要更改。

3. 针对数据结果进行统计:

最后只需要进行简单的计算即可:

1.for(re <- result ){

2.println("文档"+re._1+"开头的 文档总数:"+ re._2+",分类正确的有:"+re._3+",分类正确率是:"+(re._3*100.0/re._2)+"%")

3.}

4.val averageRate = result.map(_._3).sum *100.0 / result.map(_._2).sum

5.println("平均正确率为:"+averageRate+"%")

输出结果为:

743c8a292d6e18b4f95482b979dbb25c8e270771

4. 实验

  设置不同的numFeature,比如使用2002000,其对比结果为:

b83adff085b2529681fdc457d8a6684aebcc3c2bd2bd36d7f1756ad9f62a88d00bb12799900b2d09

所以设置numFeatures值越大,其准确率也越高,不过计算也比较复杂。

 5. 总结

1. HanLP的使用相对比较简单,这里只使用了分词及停用词,感谢开源;

2. Spark里面的TF-IDF以及Word2Vector使用比较简单,不过使用这个需要先分词;

3. 这里是在IDEA里面运行的,如果使用Spark-submit的提交方式,那么需要把hanpljar包加入,这个有待验证

相关文章
|
12月前
|
分布式计算 数据处理 Apache
Spark和Flink的区别是什么?如何选择?都应用在哪些行业?
【10月更文挑战第10天】Spark和Flink的区别是什么?如何选择?都应用在哪些行业?
1221 1
|
机器学习/深度学习 SQL 分布式计算
Spark核心原理与应用场景解析:面试经验与必备知识点解析
本文深入探讨Spark核心原理(RDD、DAG、内存计算、容错机制)和生态系统(Spark SQL、MLlib、Streaming),并分析其在大规模数据处理、机器学习及实时流处理中的应用。通过代码示例展示DataFrame操作,帮助读者准备面试,同时强调结合个人经验、行业趋势和技术发展以展现全面的技术实力。
1335 0
|
6月前
|
SQL 分布式计算 Serverless
鹰角网络:EMR Serverless Spark 在《明日方舟》游戏业务的应用
鹰角网络为应对游戏业务高频活动带来的数据潮汐、资源弹性及稳定性需求,采用阿里云 EMR Serverless Spark 替代原有架构。迁移后实现研发效率提升,支持业务快速发展、计算效率提升,增强SLA保障,稳定性提升,降低运维成本,并支撑全球化数据架构部署。
581 56
鹰角网络:EMR Serverless Spark 在《明日方舟》游戏业务的应用
|
8月前
|
机器学习/深度学习 分布式计算 大数据
阿里云 EMR Serverless Spark 在微财机器学习场景下的应用
面对机器学习场景下的训练瓶颈,微财选择基于阿里云 EMR Serverless Spark 建立数据平台。通过 EMR Serverless Spark,微财突破了单机训练使用的数据规模瓶颈,大幅提升了训练效率,解决了存算分离架构下 Shuffle 稳定性和性能困扰,为智能风控等业务提供了强有力的技术支撑。
375 15
|
8月前
|
SQL 分布式计算 Serverless
基于阿里云 EMR Serverless Spark 版快速搭建OSS日志分析应用
基于阿里云 EMR Serverless Spark 版快速搭建OSS日志分析应用
177 0
|
分布式计算 Java Serverless
EMR Serverless Spark 实践教程 | 通过 spark-submit 命令行工具提交 Spark 任务
本文以 ECS 连接 EMR Serverless Spark 为例,介绍如何通过 EMR Serverless spark-submit 命令行工具进行 Spark 任务开发。
769 7
EMR Serverless Spark 实践教程 | 通过 spark-submit 命令行工具提交 Spark 任务
|
分布式计算 运维 Serverless
EMR Serverless Spark 实践教程 | 通过 EMR Serverless Spark 提交 PySpark 流任务
在大数据快速发展的时代,流式处理技术对于实时数据分析至关重要。EMR Serverless Spark提供了一个强大而可扩展的平台,它不仅简化了实时数据处理流程,还免去了服务器管理的烦恼,提升了效率。本文将指导您使用EMR Serverless Spark提交PySpark流式任务,展示其在流处理方面的易用性和可运维性。
506 7
EMR Serverless Spark 实践教程 | 通过 EMR Serverless Spark 提交 PySpark 流任务
|
分布式计算 大数据 数据处理
Apache Spark的应用与优势:解锁大数据处理的无限潜能
【8月更文挑战第23天】Apache Spark以其卓越的性能、易用性、通用性、弹性与可扩展性以及丰富的生态系统,在大数据处理领域展现出了强大的竞争力和广泛的应用前景。随着大数据技术的不断发展和普及,Spark必将成为企业实现数字化转型和业务创新的重要工具。未来,我们有理由相信,Spark将继续引领大数据处理技术的发展潮流,为企业创造更大的价值。
|
分布式计算 大数据 Spark
Spark大数据处理:技术、应用与性能优化(全)PDF书籍推荐分享
《Spark大数据处理:技术、应用与性能优化》深入浅出介绍Spark核心,涵盖部署、实战与性能调优,适合初学者。作者基于微软和IBM经验,解析Spark工作机制,探讨BDAS生态,提供实践案例,助力快速掌握。书中亦讨论性能优化策略。[PDF下载链接](https://zhangfeidezhu.com/?p=347)。![Spark Web UI](https://img-blog.csdnimg.cn/direct/16aaadbb4e13410f8cb2727c3786cc9e.png#pic_center)
320 1
Spark大数据处理:技术、应用与性能优化(全)PDF书籍推荐分享
|
分布式计算 资源调度 测试技术
“Spark Streaming异常处理秘籍:揭秘如何驯服实时数据流的猛兽,守护你的应用稳如泰山,不容错过!”
【8月更文挑战第7天】Spark Streaming 是 Apache Spark 中的关键组件,用于实时数据流处理。部署时可能遭遇数据问题、资源限制或逻辑错误等异常。合理处理这些异常对于保持应用稳定性至关重要。基础在于理解其异常处理机制,通过 DSC 将数据流切分为 RDD。对于数据异常,可采用 try-catch 结构捕获并处理;资源层面异常需优化 Spark 配置,如调整内存分配;逻辑异常则需加强单元测试及集成测试。结合监控工具,可全面提升应用的健壮性和可靠性。
202 3

热门文章

最新文章