美文网首页
Scala-单词计数程序、并行计算,文件IO

Scala-单词计数程序、并行计算,文件IO

作者: 数据萌新 | 来源:发表于2018-10-10 16:51 被阅读0次

    hadoop和strom都有介绍过怎么进行单词计数,这里使用Scala来实现个简易的单词计数程序,在这之前补充几个常用方法

    1、排序
    排序在前面有介绍过,这里是因为单词计数用到了排序就带过一下

    scala> val lst = List(2,3,1,5,7,6,4,9,8)
    lst: List[Int] = List(2, 3, 1, 5, 7, 6, 4, 9, 8)
    
    scala> val lst2 = lst.sorted
    lst2: List[Int] = List(1, 2, 3, 4, 5, 6, 7, 8, 9)
    
    scala> lst
    res0: List[Int] = List(2, 3, 1, 5, 7, 6, 4, 9, 8)
    
    scala> lst2
    res1: List[Int] = List(1, 2, 3, 4, 5, 6, 7, 8, 9)
    

    2、分组
    使用grouped(n),每n个为一组,最后不足的略过

    scala> val lst3 = lst2.grouped(5)
    lst3: Iterator[List[Int]] = non-empty iterator
    
    scala> lst3
    res2: Iterator[List[Int]] = non-empty iterator
    
    //将Iterator转为List
    scala> val lst4 = lst3.toList
    lst4: List[List[Int]] = List(List(1, 2, 3, 4, 5), List(6, 7, 8, 9))
    
    scala> lst4
    res3: List[List[Int]] = List(List(1, 2, 3, 4, 5), List(6, 7, 8, 9))
    

    3、平铺
    flattern

    scala> val lst5 = lst4.flatten
    lst5: List[Int] = List(1, 2, 3, 4, 5, 6, 7, 8, 9)
    
    scala> lst5
    res4: List[Int] = List(1, 2, 3, 4, 5, 6, 7, 8, 9)
    

    介绍几个日常工作必备的 Scala 集合函数,如转换函数和聚合函数。

    1 最大值和最小值

    我们先从动作函数开始。

    在序列中查找最大或最小值是一个极常见的需求,较常用于面试问题和算法。还记得 Java 中的代码行吗?如下:

    int[] arr = {11, 2, 5, 1, 6, 3, 9};
     int to = arr.length - 1; 
    int max = arr[0]; 
    for (int i = 0; i < to; i++) {
     if (max < arr[i+1])
          max = arr[i+1]; 
    } 
    System.out.println(max);
    

    问题:怎么在 List 中找到最大/最小值呢?

    Scala 推荐了一个很赞的解决方案:

    val numbers = Seq(11, 2, 5, 1, 6, 3, 9)   
    numbers.max //11  
    numbers.min //1
    

    但实际操作的数据更加复杂。下面我们介绍一个更高级的例子,其中包含一个书的序列。

    case class Book(title: String, pages: Int) 
      val books = Seq( Book("Future of Scala developers", 85), 
                      Book("Parallel algorithms", 240), 
                      Book("Object Oriented Programming", 130), 
                      Book("Mobile Development", 495) ) 
      //Book(Mobile Development,495) 
      books.maxBy(book => book.pages) 
      //Book(Future of Scala developers,85) 
      books.minBy(book => book.pages)
    
    

    如上所示,minBy & maxBy 方法解决了复杂数据的问题。你只需选择决定数据最大或最小的属性。

    #2 过滤

    你过滤过集合吗?比如,筛选价格大于10美元的条目,或挑选年龄在24岁以下员工等,所有这些操作属于过滤。

    让我们举例说明:过滤一个数字 List,只获取奇数的元素。

    val numbers = Seq(1,2,3,4,5,6,7,8,9,10)
    numbers.filter(n => n % 2 == 0)
    

    然后加大难度,我想获取页数大于120页的书。

    val books = Seq(
      Book("Future of Scala developers", 85),
      Book("Parallel algorithms", 240),
      Book("Object Oriented Programming", 130),
      Book("Mobile Development", 495)
    )
      
    books.filter(book => book.pages >= 120)
    

    实际上,过滤是一个转换类型的方法,但是比运用 min 和 max 方法简单。

    还有一个与 filter 类似的方法是 filterNot。它的名字就体现了它的作用。如果你还是不了解它的实际用途,你可以在一个示例中,用 filterNot 替换 filter 方法。

    #3 Flatten

    我想大多数朋友都没听说过这个功能。其实它很好理解,我们来举例说明:

    val abcd = Seq('a', 'b', 'c', 'd') 
       val efgj = Seq('e', 'f', 'g', 'h') 
       val ijkl = Seq('i', 'j', 'k', 'l') 
       val mnop = Seq('m', 'n', 'o', 'p') 
       val qrst = Seq('q', 'r', 's', 't') 
       val uvwx = Seq('u', 'v', 'w', 'x') 
       val yz = Seq('y', 'z') 
       val alphabet = Seq(abcd, efgj, ijkl, mnop, qrst, uvwx, yz) 
       // 
      // List(a, b, c, d, e, f, g, h, i, j, k, l, m, n, o, p, q, r, s, t,
      //      u, v, w, x, y, z) 
       alphabet.flatten
    
    

    当有一个集合的集合,然后你想对这些集合的所有元素进行操作时,就会用到 flatten。

    #4 Euler Diagram 函数

    不要紧张!接下来的操作大家都熟知:差集、交集和并集。以下示例能很好地解释 Euler Diagram 函数:

    val num1 = Seq(1, 2, 3, 4, 5, 6)
    val num2 = Seq(4, 5, 6, 7, 8, 9)
      
    //List(1, 2, 3)
    num1.diff(num2)
      
    //List(4, 5, 6)
    num1.intersect(num2)
      
    //List(1, 2, 3, 4, 5, 6, 4, 5, 6, 7, 8, 9)
    num1.union(num2)
    

    上述示例中的 union 保留了重复的元素。如果我们不需要重复怎么办?这时可以使用 distinct 函数:

    /List(1, 2, 3, 4, 5, 6, 7, 8, 9)
    num1.union(num2).distinct
    

    下面是上述功能的图示:

    image.png

    #5 map(映射)列表元素

    map 是 Scala 集合最常用的一个函数。它的功能十分强大:

    val numbers = Seq(1,2,3,4,5,6)
      
    //List(2, 4, 6, 8, 10, 12)
    numbers.map(n => n * 2)
      
    val chars = Seq('a', 'b', 'c', 'd')
      
    //List(A, B, C, D)
    chars.map(ch => ch.toUpper)
    

    map 函数的逻辑是遍历集合中的元素并对每个元素调用函数。你也可以不调用任何函数,保持返回元素本身,但这样 map 无法发挥作用,因为你在映射过后得到的是同样的集合。

    #6 flatMap

    我很难具体说明 flatMap 的使用场合,因为很多不同的情况下都会用到 flatMap。如果大家仔细观察,就会发现 flatMap 是由下列这两个函数组成的:

    map & flatten

    现在,假设我们想知道字母表中的大写字母和小写字母的排列情况:

    val abcd = Seq('a', 'b', 'c', 'd')
      
    //List(A, a, B, b, C, c, D, d)
    abcd.flatMap(ch => List(ch.toUpper, ch))
    
    scala> val line = "this is a book"
    line: String = this is a book
    
    scala> val words = line.split(" ")
    words: Array[String] = Array(this, is, a, book)
    
    scala> val arrayofChars = words.flatMap(_.toList)
    arrayofChars: Array[Char] = Array(t, h, i, s, i, s, a, b, o, o, k)
    

    #7 对整个集合进行条件检查

    有一个场景大家都知道,即确保集合中所有元素都要符合某些要求,如果有哪怕一个元素不符合条件,就需要进行一些处理:

    val numbers = Seq(3, 7, 2, 9, 6, 5, 1, 4, 2)
      
    //ture
    numbers.forall(n => n < 10)
      
    //false
    numbers.forall(n => n > 5)
    

    而 forall 函数就是为处理这类需求而创建的。

    #8 对集合进行分组

    你是否尝试过将一个集合按一定的规则拆分成两个新的集合?比如,我们把某个集合拆分成偶数集和奇数集,partition 函数可以帮我们做到这一点:

    val numbers = Seq(3, 7, 2, 9, 6, 5, 1, 4, 2)    
    //(List(2, 6, 4, 2), List(3, 7, 9, 5, 1))    
    numbers.partition(n => n % 2 == 0)
    

    #9 Fold

    另一个流行的操作是 fold。 在 Scala 的上下文中,通常可以考虑 foldLeft 和 foldRight。他们是从不同的方面做同样的工作:

    val numbers = Seq(1, 2, 3, 4, 5)
      
    //15
    numbers.foldLeft(0)((res, n) => res + n)
    

    在第一对括号中,我们放一个起始值。 在第二对括号中,我们定义需要对数字序列的每个元素执行的操作。 第一步,n = 0,然后它根据序列元素变化。

    另一个关于 foldLeft 的例子,计算字符数:

    val words = Seq("apple", "dog", "table")    
    //13
     words.foldLeft(0)((resultLength, word) => resultLength + word.length)
    

    foreach

    foreach 方法的参数是一个函数,它把这个函数作用于集合中的每一个元素,但是不返回任何东西。它和 map 类似,唯一的区别在于 map 返回一个集合,而 foreach 不返回任何东西。由于它的无返回值特性它很少使用。

    scala> val words = "scala is fun".split(" ")
    words: Array[String] = Array(scala, is, fun)
    
    scala> words.foreach(println)
    scala
    is
    fun
    

    reduce

    reduce 方法返回一个值。顾名思义,它将一个集合整合成一个值。它的参数是一个函数,这个函数有两个参数,并返回一个值。从本质上说,这个函数是一个二元操作符,并且满足结合律和交换律。

    scala> val xs = List(2,3,4,5,6)
    xs: List[Int] = List(2, 3, 4, 5, 6)
    
    scala> var sum = xs.reduce(_+_)
    sum: Int = 20
    
    scala> var sum = xs.reduce(_-_)
    sum: Int = -16
    
    scala> var sum = xs.reduce(_/_)
    sum: Int = 0
    
    scala> var sum = xs.reduce(_*_)
    sum: Int = 720
    
    

    下面是一个找出句子中最长单词的例子

    scala> val words = "scala is fun".split(" ")
    words: Array[String] = Array(scala, is, fun)
    
    scala> val longestWord = words.reduce((w1,w2)=>if(w1.length > w2.length) w1 else
     w2)
    longestWord: String = scala
    

    wordcount程序编写

    注:思路是跟mapreduce做单词统计一样而不是跟常规的java编写一样。

    //对三个字符串中的单词做单词数量统计
    scala> val lines = List("hadoop hdfs mr hive","hdfs hive hbase storm kafka","hive hbase storm kafka spark")
    lines: List[String] = List(hadoop hdfs mr hive, hdfs hive hbase storm kafka, hive hbase storm kafka spark)
    
    //3个字符串中的单词均根据空格切割
    scala> lines.map(_.split(" "))
    res18: List[Array[String]] = List(Array(hadoop, hdfs, mr, hive), Array(hdfs, hive, hbase, storm, kafka), Array(hive, hbase, storm, kafka, spark))
    
    //平铺
    scala> lines.map(_.split(" ")).flatten
    res19: List[String] = List(hadoop, hdfs, mr, hive, hdfs, hive, hbase, storm, kafka, hive, hbase, storm, kafka, spark)
    
    scala> val words = lines.map(_.split(" ")).flatten
    words: List[String] = List(hadoop, hdfs, mr, hive, hdfs, hive, hbase, storm, kafka, hive, hbase, storm, kafka, spark)
    
    //每个元素都写为(单词,1)的对偶格式
    scala> words.map((_,1))
    res20: List[(String, Int)] = List((hadoop,1), (hdfs,1), (mr,1), (hive,1), (hdfs,1), (hive,1), (hbase,1), (storm,1), (kafka,1), (hive,1), (hbase,1), (storm,1), (kafka,1), (spark,1))
    
    根据单词进行分组
    scala> words.map((_,1)).groupBy(_._1)
    res21: scala.collection.immutable.Map[String,List[(String, Int)]] = Map(storm -> List((storm,1), (storm,1)), kafka -> List((kafka,1), (kafka,1)), hadoop -> List((hadoop,1)), spark -> List((spark,1)), hive -> List((hive,1), (hive,1), (hive,1)), mr -> List((mr,1)), hbase -> List((hbase,1), (hbase,1)), hdfs -> List((hdfs,1), (hdfs,1)))
    
    scala> val grouped = words.map((_,1)).groupBy(_._1)
    grouped: scala.collection.immutable.Map[String,List[(String, Int)]] = Map(storm -> List((storm,1), (storm,1)), kafka -> List((kafka,1), (kafka,1)), hadoop -> List((hadoop,1)), spark -> List((spark,1)), hive -> List((hive,1), (hive,1), (hive,1)), mr -> List((mr,1)), hbase -> List((hbase,1), (hbase,1)), hdfs -> List((hdfs,1), (hdfs,1)))
    
    scala> grouped.map(t => (t._1,t._2.size))
    res22: scala.collection.immutable.Map[String,Int] = Map(storm -> 2, kafka -> 2, hadoop -> 1, spark -> 1, hive -> 3, mr -> 1, hbase -> 2, hdfs -> 2)
    
    //单词统计和排序,但是必须转List才能使用sortBy
    scala> grouped.map(t => (t._1,t._2.size)).sortedBy(_._2)
    <console>:14: error: value sortedBy is not a member of scala.collection.immutable.Map[String,Int]
                  grouped.map(t => (t._1,t._2.size)).sortedBy(_._2)
                                                     ^
    
    scala> grouped.map(t => (t._1,t._2.size)).toList
    res25: List[(String, Int)] = List((storm,2), (kafka,2), (hadoop,1), (spark,1), (hive,3), (mr,1), (hbase,2), (hdfs,2))
    
    scala> grouped.map(t => (t._1,t._2.size)).toList.sortBy(_._2)
    res26: List[(String, Int)] = List((hadoop,1), (spark,1), (mr,1), (storm,2), (kafka,2), (hbase,2), (hdfs,2), (hive,3))
    
    scala> val result = grouped.map(t => (t._1,t._2.size)).toList.sortBy(_._2)
    result: List[(String, Int)] = List((hadoop,1), (spark,1), (mr,1), (storm,2), (kafka,2), (hbase,2), (hdfs,2), (hive,3))
    

    并行计算的一些方法

    scala> val a = Array(1,2,3,4,5,6)
    a: Array[Int] = Array(1, 2, 3, 4, 5, 6)
    
    scala> a.sum
    res41: Int = 21
    
    scala> a.reduce(_+_)  //reduce调的是reduceLeft,从左往右操作
    res42: Int = 21
    
    scala> a.reduce(_-_)
    res43: Int = -19
    
    
    scala> a.par //转换为并行化集合
    res44: scala.collection.parallel.mutable.ParArray[Int] = ParArray(1, 2, 3, 4, 5, 6)
    
    scala> a.par.reduce(_+_)//会将集合切分为好几块然后并行计算最后汇总
    res45: Int = 21
    
    scala> a.fold(10)(_+_)  //先给初始值10,加上a中所有值之和。fold是并行计算,第一个_表示初始值或者累加过后的结果
    res46: Int = 31
    
    scala> a.par.fold(10)(_+_) //并行化之后可能就不一样了,每份都要+10
    res47: Int = 51
    
    scala> a.par.fold(0)(_+_)
    res49: Int = 21
    

    文件I/O

    Scala 进行文件写操作,直接用的都是 java中 的 I/O 类 (java.io.File):

    import java.io._
    
    object Test {
       def main(args: Array[String]) {
          val writer = new PrintWriter(new File("test.txt" ))
    
          writer.write("hello world")
          writer.close()
       }
    }
    

    从屏幕读取输入

    object Test {
       def main(args: Array[String]) {
          print("请输入你的名字 : " )
          val line = Console.readLine
          
          println("谢谢,你输入的是: " + line)
       }
    }
    

    从文件上读取内容
    从文件读取内容非常简单。我们可以使用 Scala 的 Source 类及伴生对象来读取文件。

    import scala.io.Source
    
    object Test {
       def main(args: Array[String]) {
          println("文件内容为:" )
    
          Source.fromFile("test.txt" ).foreach{ 
             print 
          }
       }
    }
    
    import scala.io.Source
    
    object Test {
       def main(args: Array[String]) {
          println("文件内容为:" )
    
          Source.fromFile("test.txt" ).getLines.foreach{ 
             print 
          }
       }
    }
    

    相关文章

      网友评论

          本文标题:Scala-单词计数程序、并行计算,文件IO

          本文链接:https://www.haomeiwen.com/subject/urqmaftx.html