2-1、Spark的单词统计WC

不念不忘少年蓝@ 2022-08-18 11:32 200阅读 0赞

1、输入数据：

[root@spark0 bigdata]# pwd
    /usr/local/spark-1.5.2-bin-hadoop2.6/bigdata
    [root@spark0 bigdata]# more wcDemo1.txt 
    hadoop  hive
    solr    redis
    kafka   hadoop
    storm   flume
    sqoop   docker
    spark   spark
    hadoop  spark
    elasticsearch   hbase
    hadoop  hive
    spark   hive
    hadoop  spark
    [root@spark0 bigdata]#

2、单词计数：

scala> val rdd=sc.textFile("/usr/local/spark-1.5.2-bin-hadoop2.6/bigdata/wcDemo1.txt").flatMap(_.split("\t")).map(x=>(x,1)).reduceByKey(_+_).collect
    rdd: Array[(String, Int)] = Array((spark,5), (hive,3), (hadoop,5), (docker,1), (flume,1), (solr,1), (storm,1), (elasticsearch,1), (kafka,1), (sqoop,1), (redis,1), (hbase,1))

3、对结果升序排序：

scala> val rdd=sc.textFile("/usr/local/spark-1.5.2-bin-hadoop2.6/bigdata/wcDemo1.txt").flatMap(_.split("\t")).map(x=>(x,1)).reduceByKey(_+_).sortByKey().collect
    rdd: Array[(String, Int)] = Array((docker,1), (elasticsearch,1), (flume,1), (hadoop,5), (hbase,1), (hive,3), (kafka,1), (redis,1), (solr,1), (spark,5), (sqoop,1), (storm,1))

4、对结果降序排序：

scala> val rdd=sc.textFile("/usr/local/spark-1.5.2-bin-hadoop2.6/bigdata/wcDemo1.txt").flatMap(_.split("\t")).map(x=>(x,1)).reduceByKey(_+_).sortByKey(false).collect
    rdd: Array[(String, Int)] = Array((storm,1), (sqoop,1), (spark,5), (solr,1), (redis,1), (kafka,1), (hive,3), (hbase,1), (hadoop,5), (flume,1), (elasticsearch,1), (docker,1))

5、统计结果行数

scala> val rdd=sc.textFile("/usr/local/spark-1.5.2-bin-hadoop2.6/bigdata/wcDemo1.txt").flatMap(_.split("\t")).map(x=>(x,1)).reduceByKey(_+_).sortByKey(false).count
    rdd: Long = 12

6、将结果进行保存：

scala> val rdd=sc.textFile("/usr/local/spark-1.5.2-bin-hadoop2.6/bigdata/wcDemo1.txt").flatMap(_.split("\t")).map(x=>(x,1)).reduceByKey(_+_).sortByKey(false).saveAsTextFile("/usr/local/spark-1.5.2-bin-hadoop2.6/bigdata/wcDemo_out")
    rdd: Unit = ()
    
    
    scala>

7、解析：  
对于WC而言，需要从输入数据中每行字符串解析出单词，然后将相同的单词放到一个桶中，最后统计每个桶中每个单词出现的频率。  
  
  
flatMap函数将一条记录转换成多条记录（一对多关系），map函数将一条记录转换成另一条记录（一对一关系），reduceByKey函数将Key相同的数据划分到一个桶中，并以Key为单位分组进行计算。  
  
  
经过一系列的RDD转换算子操作，之前都是Transformation算子，最后collect、saveAsTextFile、count都是Actions算子。  
  
  
  
8、查看结果

[root@spark0 wcDemo_out]# pwd
    /usr/local/spark-1.5.2-bin-hadoop2.6/bigdata/wcDemo_out
    [root@spark0 bigdata]# cd wcDemo_out/
    [root@spark0 wcDemo_out]# ll
    总用量 4
    -rw-r--r--. 1 root root 128 12月 13 22:35 part-00000
    -rw-r--r--. 1 root root   0 12月 13 22:35 _SUCCESS
    [root@spark0 wcDemo_out]# pwd
    /usr/local/spark-1.5.2-bin-hadoop2.6/bigdata/wcDemo_out
    [root@spark0 wcDemo_out]# ll
    总用量 4
    -rw-r--r--. 1 root root 128 12月 13 22:35 part-00000
    -rw-r--r--. 1 root root   0 12月 13 22:35 _SUCCESS
    [root@spark0 wcDemo_out]# more part-00000 
    (storm,1)
    (sqoop,1)
    (spark,5)
    (solr,1)
    (redis,1)
    (kafka,1)
    (hive,3)
    (hbase,1)
    (hadoop,5)
    (flume,1)
    (elasticsearch,1)
    (docker,1)
    [root@spark0 wcDemo_out]#