scala:word count 单词计数

创建maven工程，导入jar包：

  UTF-8
  1.8
  1.8
  2.12.10
  3.1.2
  2.12
  1.2.17



  
    org.apache.spark
    spark-core_${spark.scala.version}
    ${spark.version}
  
  
    org.apache.spark
    spark-sql_${spark.scala.version}
    ${spark.version}
  
  
    org.scala-lang
    scala-library
    ${scala.version}
  
  
    log4j
    log4j
    ${log4j.version}
  

配置log4j文件：src/main/resources/log4j.properties

#指定log4j的输出信息
log4j.rootLogger=INFO, stdout, logfile
#指定log4j的标准输出
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
#指定log4j的标准输出的样式
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
#指定标准输出的转换的格式
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
#指定日志文件的输出
log4j.appender.logfile=org.apache.log4j.FileAppender
#指定log4j的输出路径文件名
log4j.appender.logfile.File=log/hd.log
#指定日志日志输出样式
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
#指定日志文件的转换格式
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

创建scala对象，编程：

import org.apache.spark.{SparkConf, SparkContext}

object App {
  def main(args: Array[String]): Unit = {
    val config: SparkConf = new SparkConf()
      .setAppName("spark_rdd_03")//设置任务名称
      .setMaster("local[*]")//设置Master,本地模式，“ * ” 表示cup资源有多少，用多少；也可以用local[2]，表示2个核
    val sc = new SparkContext(config)
    sc
      .textFile("hdfs://single01:9000/test/file/Broken to Harness.txt",3)//文件路径和分区数
        .mapPartitions(it=>{    //多分区，每个分区都是一个迭代器，每个分区单独处理（单分区可以用map）
          it    //以行迭代
            .filter(_.trim.size>0)  //清除空行
            .flatMap(   //一行有多个单词（数据），不是单一数据所以降维（如果一行作为一条数据，可以用map）
              _
                .replaceAll(",|\.|!|\?|;|"|-","")//清洗每行的标点符号
                .replaceAll("\s{2,}"," ")//两个或以上的空白字符变为一个空格
                .split(" ")//以空格分割单词
                .map((_,1))//改变结果（单词，1）
                .groupBy(_._1)//以单词分组(有点：减少后面的shuffle操作数据迁移量，类似于combiner)
                .map(tp2=>(tp2._1,tp2._2.size))//统计单词和单词数量
            )
        })
        .reduceByKey(_+_)//自动按键分组，把相邻的两个同键（单词）的值相加
        .foreach(println)
    sc.stop()
  }
}