转载

[原]Spark中文分词统计功能main方法中调用

整个工程总共包括GuiUtils.scala,SparkWordCount.scala,TextPane.scala和WordCounter.scala四个Scala类和JavaUtil.java一个Java类。

GuiUtils.scala完全复制自网上代码,实现了类似 于Java Swing中OptionPanel的message 提示框的功能。

   TextPane.scala则复制自GitHub上的ScalaSwing2项目,把JTextPanel移植到了Scala中。标准的Scala Library直到2.1.18版本都没有实现

TextPanel,只有TextArea,我们的工程中显示分词结果沿用了Java版的JTextPane,所以我们复制了这个Scala版的。

SparkWordCount.scala类实现了Spark中文分词统计的核心功能,是在DT 大数据梦工厂王家林老师的SparkWordCount的代码基础上改写的。

    首先,把主要功能步骤从伴生对象的main方法中移到了SparkWordCount类中,并拆分为多个方法,使得伴生对象的main方法和后面的GUI界面

都能调用:

class SparkWordCount {
           var sc:SparkContext = null;

           def initSpark(appName:String){
                 /**
                  * 第1步:创建Spark的配置对象SparkConf,设置Spark程序的运行时的配置信息,
                  * 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL,如果设置
                  * 为local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差(例如
                  * 只有1G的内存)的初学者       * 
                  */
                  val conf = new SparkConf() //创建SparkConf对象
                  conf.setAppName(appName) //设置应用程序的名称,在程序运行的监控界面可以看到名称
                  conf.setMaster("local") //此时,程序在本地运行,不需要安装Spark集群

                 /**
                  * 第2步:创建SparkContext对象
                  * SparkContext是Spark程序所有功能的唯一入口,无论是采用Scala、Java、Python、R等都必须有一个SparkContext
                  * SparkContext核心作用:初始化Spark应用程序运行所需要的核心组件,包括DAGScheduler、TaskScheduler、SchedulerBackend
                  * 同时还会负责Spark程序往Master注册程序等
                  * SparkContext是整个Spark应用程序中最为至关重要的一个对象
                  */
                  sc = new SparkContext(conf) //创建SparkContext对象,通过传入SparkConf实例来定制Spark运行的具体参数和配置信息
            }

            def wordCount(doc:String, wordLength:Int):RDD[(String,Int)]={
                  /**
                   * 第3步:根据具体的数据来源(HDFS、HBase、Local FS、DB、S3等)通过SparkContext来创建RDD
                   * RDD的创建基本有三种方式:根据外部的数据来源(例如HDFS)、根据Scala集合、由其它的RDD操作
                   * 数据会被RDD划分成为一系列的Partitions,分配到每个Partition的数据属于一个Task的处理范畴
                   */
                    //val lines = sc.textFile("E://text//唐诗三百首.txt", 1) //读取本地文件并设置为一个Partion
                    //val lines = sc.textFile("src/com/magicstudio/spark/text/唐诗三百首.txt", 1) 
                    val lines = sc.textFile(doc, 1) 

                    /**
                     * 第4步:对初始的RDD进行Transformation级别的处理,例如map、filter等高阶函数等的编程,来进行具体的数据计算
                     *  第4.1步:讲每一行的字符串拆分成单个的单词
                     */

                     //val words = lines.flatMap { line => line.split(" ")} //对每一行的字符串进行单词拆分并把所有行的拆分结果通过flat合并成为一个大的单词集合
                     val words = lines.flatMap { line => JavaUtil.getSplitWords(line, wordLength).asScala }
                     /**
                      * 第4步:对初始的RDD进行Transformation级别的处理,例如map、filter等高阶函数等的编程,来进行具体的数据计算
                      *  第4.2步:在单词拆分的基础上对每个单词实例计数为1,也就是word => (word, 1)
                      */
                     val pairs = words.map { word => (word, 1) }

                     /**
                      * 第4步:对初始的RDD进行Transformation级别的处理,例如map、filter等高阶函数等的编程,来进行具体的数据计算
                      *  第4.3步:在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数
                      */
                     val wordCounts = pairs.reduceByKey(_+_) //对相同的Key,进行Value的累计(包括Local和Reducer级别同时Reduce)

                      //added by Dumbbell Yang at 2016-07-24
                     wordCounts.sortBy(x => x._2 , false, wordCounts.partitions.size)
           }

           def outputResult(wordCounts:RDD[(String,Int)]){
                    wordCounts.foreach(wordNumberPair => println(wordNumberPair._1 + " : " + wordNumberPair._2))
           }

           def closeSpark(){
                   sc.stop()
            }

}

其次,在wordCount方法中,把原来第3步读取固定文件的方式改为参数方式,可以是src目录下的相对文件路径(在GUI界面上通过下拉

框选择),也可以是本地磁盘上的绝对文件路径(通过文件浏览框选择):

//val lines = sc.textFile("E://text//唐诗三百首.txt", 1) //读取本地文件并设置为一个Partion
  //val lines = sc.textFile("src/com/magicstudio/spark/text/唐诗三百首.txt", 1) 
  val lines = sc.textFile(doc, 1)

  然后就是第4.1步中,通过调用JavaUtil类中的java方法,实现了中文分词功能,替换掉原来简单的split,对每一行文本进行中文分词:

  //val words = lines.flatMap { line => line.split(" ")} //对每一行的字符串进行单词拆分并把所有行的拆分结果通过flat合并成为一个大的单词集合
  val words = lines.flatMap { line => JavaUtil.getSplitWords(line, wordLength).asScala }

  需要注意的是,由于需要调用Java功能,在Scala和Java之间进行数据传递,所以必须引用数据类型转换的library:

  import collection.JavaConverters._

  然后,才可以对JavaUtil中的getSplitWords方法返回的结果进行asScala的转换,使之能够满足Scala方法调用的要求。

  最后的一个改动,就是加上了一个对分词统计结果按照词频进行排序的功能:

  //added by Dumbbell Yang at 2016-07-24
  wordCounts.sortBy(x => x._2 , false, wordCounts.partitions.size)

  可以对比Java方法实现排序时,交换key和value,排序,然后在交换回去的繁琐,scala语言确实方便很多。
原文  http://blog.csdn.net/xieyuooo/article/details/52270100
正文到此结束
Loading...