基于Spark的中文文章词频统计

1.中文分词

中文分词选用的是Ansj框架。分词部分使用Java语言封装，由Spark中的flatMap算子调用。分词后根据词性做了筛选。

package com.ahn.spark;import org.ansj.domain.Result;import org.ansj.domain.Term;import org.ansj.splitWord.analysis.ToAnalysis;import java.util.*;public class AnsjTest { public static List fenci(String str){ //筛选词性集合 Set expectedNature = new HashSet() {{ add("n"); add("v"); add("vd"); add("vn"); add("vf"); add("vx"); add("vi"); add("vl"); add("vg"); add("nt"); add("nz"); add("nw"); add("nl"); add("ng"); add("userDefine"); add("wh"); }}; //存放结果的List List array=new ArrayList<>(); //开始分词 Result result = ToAnalysis.parse(str); //得到结果集 List terms = result.getTerms(); terms.forEach(t->{ //根据筛选集合筛选出符合条件的数据 if (expectedNature.contains(t.getNatureStr())) { array.add(t.getName()); } }); return array; }}

2.词频统计

词频统计使用Scala进行编写，具体步骤与WorldCount相同。

package com.ahn.sparkimport org.apache.spark.{SparkConf, SparkContext}import com.ahn.spark.AnsjTest.{fenci}object Test { def main(args: Array[String]): Unit = { //初始化SparkConf val conf: SparkConf = new SparkConf() //设置使用本地运行的方式 还可以选择Spark自身集群的方式 以及on Yarn的方式 conf.setMaster("local") //设置应用名称 conf.setAppName("SparkTest") //初始化SparkContext val context = new SparkContext(conf) //设置日志打印级别 context.setLogLevel("ERROR") //读取数据文件 val value = context.textFile("src/main/resources/test.txt") //flatMap算子中使用fenci()方法 val words = value.flatMap(v => fenci(v).toArray()) //Map算子将数据处理为(key 1)的形式 val pairs = words.map(word => (word, 1)) //reduceByKey根据key进行合并 将value的值进行相加。 val wordcounts = pairs.reduceByKey(_ + _) //根据value值进行降序排序 并取出前10条记录打印 wordcounts.sortBy(x=>x._2,false).take(10).foreach(println) }}

3.pom文件

<?xml version="1.0" encoding="UTF-8"?> 4.0.0 org.example spark_test 1.0-SNAPSHOT 8 8 aliyun http://maven.aliyun.com/nexus/content/groups/public/ cloudera https://repository.cloudera.com/artifactory/cloudera-repos/ jboss http://repository.jboss.com/nexus/content/groups/public org.apache.spark spark-core_2.12 3.0.1 org.apache.spark spark-sql_2.12 3.0.1 provided org.ansj ansj_seg 5.1.6

4.项目目录

5.结果