文章目錄

• Spark WordCount 案例
• • 1、程序連接 Spark
  • 2、WordCount 案例示例
  • 3、復(fù)雜版 WordCount
  • 4、Spark 框架Wordcount

10年積累的成都網(wǎng)站制作、成都網(wǎng)站建設(shè)經(jīng)驗(yàn)，可以快速應(yīng)對(duì)客戶對(duì)網(wǎng)站的新想法和需求。提供各種問(wèn)題對(duì)應(yīng)的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡(luò)服務(wù)。我雖然不認(rèn)識(shí)你，你也不認(rèn)識(shí)我。但先網(wǎng)站設(shè)計(jì)后付款的網(wǎng)站建設(shè)流程，更有盈江免費(fèi)網(wǎng)站建設(shè)讓你可以放心的選擇與我們合作。Spark WordCount 案例 1、程序連接 Spark

首先這個(gè)Scala spark程序和spark的鏈接，跟sql編程類似。首先new 一個(gè)新的val context = SparkContext()對(duì)象，然后還要用到
val conf = SparkConf.setMaster("local").setAppName("WordCount")這個(gè)是配置信息，比如這個(gè)是本地連接所以里面是local，然后后面那個(gè)是程序的名字，這個(gè)寫(xiě)完之后，吧這個(gè)conf對(duì)象放在SparkContext(conf)這里面。然后在程序的最后，用完了要關(guān)閉連接，context.stop()，使用stop方法關(guān)閉

2、WordCount 案例示例

先在D盤(pán)，把要測(cè)試的文件數(shù)據(jù)準(zhǔn)備好

思路：首先連接之后，第一步是讀取文件,使用textFile()方法，里面的參數(shù)是要讀取的文件的路徑，然后把文件一行一行的讀取出來(lái)。第二步是使用flatMap(_.split(" "))方法，進(jìn)行map映射和扁平化，把單詞按照空格分割開(kāi)。第三步是groupBy(word =>word)按照單詞進(jìn)行分組，一樣的單詞分到一組。第四步map()映射進(jìn)行模式匹配，取去key和他的集合的size也就是單詞出現(xiàn)的次數(shù)。然后使用collect()方法將結(jié)果采集打印，最后使用foreach(println)進(jìn)行遍歷。

package com.atguigu.bigdata.spark.core.wc

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

class spark01_WordCount {}
object spark01_WordCount{def main(args: Array[String]): Unit = {// Application 我們自己寫(xiě)的應(yīng)用程序
    // Spark 框架
    //用我們的應(yīng)用程序去連接spark 就跟那個(gè)sql 編程一樣
    //TODD建立和Spark 框架的連接
     //1、Java里面是Conntection 進(jìn)行連接
     //2、Scala 里有個(gè)類似的，SparkContext()
      //2.1 SparkConf()配置不然不曉得連的哪個(gè). setMaster() 里面是本地連接,setAppName() 里面是app的名稱
    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val context = new SparkContext(sparkConf)
    println(context)

    //TODD 執(zhí)行業(yè)務(wù)操作
    //1、讀取文件,獲取一行一行的數(shù)據(jù) 這一步是扁平化
    //hello word
    val value = context.textFile("D:\\wc.txt") //textFile 可以吧文件一行一行的讀出來(lái)


    //2、將數(shù)據(jù)進(jìn)行拆分，形成一個(gè)一個(gè)的單詞
    //扁平化：將整體拆分為個(gè)體的操作
    //"hello word" =>hello，word
    val danci: RDD[String] = value.flatMap(a =>a.split(" ")) //根據(jù)空格進(jìn)行拆分


    //3、將數(shù)據(jù)根據(jù)單詞進(jìn)行分組，便于統(tǒng)計(jì)
    //(hello,hello,hello,hello,hello),(word,word,word) 這個(gè)樣子的
    //按照單詞進(jìn)行分組
    val wordGroup = danci.groupBy(word =>word) //按照單詞進(jìn)行分組

    //4、對(duì)分組數(shù)的數(shù)據(jù)進(jìn)行轉(zhuǎn)換
    //(hello,hello,hello,hello,hello),(word,word,word)
    //(hello,5),(word,3)
    val wordToCount =  wordGroup.map{//模式匹配
      case (word,list) =>{(word,list.size) //匹配，第一個(gè)是單詞。第二個(gè)是長(zhǎng)度，這個(gè)長(zhǎng)度就是單詞出現(xiàn)的次數(shù)
      }
    }

    //5、將轉(zhuǎn)換結(jié)果采集到控制臺(tái)打印出來(lái)
    val tuples = wordToCount.collect() //collect()方法，將結(jié)果采集打印
    tuples.foreach(println)



    //TODD 關(guān)閉連接
    context.stop() //這樣就關(guān)閉連接了


  }
}

3、復(fù)雜版 WordCount

因?yàn)橹澳莻€(gè)是用size方法得到次數(shù)，但是這樣就不像是一個(gè)聚合操作，所以使用map映射，然后使用reduce 進(jìn)行聚合操作，這樣來(lái)得到單詞出現(xiàn)的次數(shù)。

package com.atguigu.bigdata.spark.core.wc

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

//復(fù)雜版wordcount
class spark01_fuzaWrodCount {}
object spark01_fuzaWrodCount{def main(args: Array[String]): Unit = {//之前是使用size 方法，得出單詞出現(xiàn)的次數(shù)，但是那樣實(shí)現(xiàn)不像是個(gè)聚合功能，所以我們改善一下
    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val context = new SparkContext(sparkConf)
    println(context)

    //TODD 執(zhí)行業(yè)務(wù)操作
    //1、讀取文件,獲取一行一行的數(shù)據(jù) 這一步是扁平化
    //hello word
    val value = context.textFile("D:\\wc.txt") //textFile 可以吧文件一行一行的讀出來(lái)


    //2、將數(shù)據(jù)進(jìn)行拆分，形成一個(gè)一個(gè)的單詞
    //扁平化：將整體拆分為個(gè)體的操作
    //"hello word" =>hello，word
    val danci: RDD[String] = value.flatMap(a =>a.split(" ")) //根據(jù)空格進(jìn)行拆分

    val wordToOne: RDD[(String, Int)] = danci.map(word =>(word, 1)) //直接在這一步統(tǒng)計(jì)單詞出現(xiàn)的次數(shù)
    val wordGroup: RDD[(String, Iterable[(String, Int)])] = wordToOne.groupBy(t =>t._1) //然后按照方式，取第一個(gè)元素為分組的依據(jù)
    val wordToCount = wordGroup.map{//這一步不是用size了
      case (word,list) =>{list.reduce(
          (t1,t2) =>{(t1._1,t1._2 + t2._2)
          }
        )
      }
    }
    //這里不是直接size，而是進(jìn)行reduce，聚合操作，將key給加起來(lái)
    //val wordCount2 = wordGroup.map{case (word,list)=>{ list.reduce((t1,t2)=>{(t1._1,t1._2+t2._2)})}}
    val array: Array[(String, Int)] = wordToCount.collect() //采集結(jié)果打印輸出
    array.foreach(println) //foreach()方法進(jìn)行遍歷
    

    //TODD 關(guān)閉連接
    context.stop() //這樣就關(guān)閉連接了


  }
}

4、Spark 框架Wordcount

Spark框架里面有個(gè)方法，分組和聚合可以一個(gè)方法完成reduceByKey(_ + _)，這樣大大減少了代碼量，從讀取文件進(jìn)來(lái)，到輸出結(jié)果四五行就能完成這個(gè)案例。

package com.atguigu.bigdata.spark.core.wc

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

//使用saprk框架進(jìn)行統(tǒng)計(jì)
class spark02_sparkCount {}
object spark02_sparkCount{def main(args: Array[String]): Unit = {//之前是使用size 方法，得出單詞出現(xiàn)的次數(shù)，但是那樣實(shí)現(xiàn)不像是個(gè)聚合功能，所以我們改善一下
    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val context = new SparkContext(sparkConf)
    println(context)

    //TODD 執(zhí)行業(yè)務(wù)操作
    //1、讀取文件,獲取一行一行的數(shù)據(jù) 這一步是扁平化
    //hello word
    val value = context.textFile("D:\\wc.txt") //textFile 可以吧文件一行一行的讀出來(lái)


    //2、將數(shù)據(jù)進(jìn)行拆分，形成一個(gè)一個(gè)的單詞
    //扁平化：將整體拆分為個(gè)體的操作
    //"hello word" =>hello，word
    val danci: RDD[String] = value.flatMap(a =>a.split(" ")) //根據(jù)空格進(jìn)行拆分

    val wordToOne: RDD[(String, Int)] = danci.map(word =>(word, 1)) //直接在這一步統(tǒng)計(jì)單詞出現(xiàn)的次數(shù)

    //Spark 框架提供了更多的功能，可以將分組和聚合使用一個(gè)功能實(shí)現(xiàn)
    //reduceByKey():相同的key的數(shù)據(jù)，可以對(duì)value進(jìn)行reduce聚合 這是spark提供的功能
    val wordCount = wordToOne.reduceByKey((x,y) =>x+y) //相當(dāng)于同一個(gè)key 進(jìn)行累加_ + _ 可以簡(jiǎn)化成這樣


    val array: Array[(String, Int)] = wordCount.collect() //采集結(jié)果打印輸出
    array.foreach(println) //foreach()方法進(jìn)行遍歷


    //TODD 關(guān)閉連接
    context.stop() //這樣就關(guān)閉連接了


  }
}

簡(jiǎn)化下來(lái)就是這幾步

你是否還在尋找穩(wěn)定的海外服務(wù)器提供商？創(chuàng)新互聯(lián)www.cdcxhl.cn海外機(jī)房具備T級(jí)流量清洗系統(tǒng)配攻擊溯源，準(zhǔn)確流量調(diào)度確保服務(wù)器高可用性，企業(yè)級(jí)服務(wù)器適合批量采購(gòu)，新人活動(dòng)首月15元起，快前往官網(wǎng)查看詳情吧

網(wǎng)站題目：SparkWordCount案例-創(chuàng)新互聯(lián)
網(wǎng)頁(yè)網(wǎng)址：http://www.js-pz168.com/article8/dgdcop.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站內(nèi)鏈、用戶體驗(yàn)、響應(yīng)式網(wǎng)站、網(wǎng)頁(yè)設(shè)計(jì)公司、動(dòng)態(tài)網(wǎng)站、品牌網(wǎng)站制作

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)