Spark作業(yè)與MapReduce作業(yè)同樣可以先在本地開發(fā)測試，本地執(zhí)行模式與集群提交模式，代碼的業(yè)務功能相同，因此本書大多數采用本地開發(fā)模式。下面講解使用IDEA工具開發(fā)WordCount單詞計數程序的相關步驟。

　　1.創(chuàng)建Maven項目，新建資源文件夾

　　創(chuàng)建一個Maven工程項目，名為“spark_chapter02”。項目創(chuàng)建好后，在main和test目錄下分別創(chuàng)建一個名稱為scala的文件夾，創(chuàng)建好的目錄結構如圖1所示。

　圖1 Spark_chapter02項目目錄結構

　　在圖1中，選中main目錄下的scala文件夾，單擊右鍵選擇【Mark Directory as】→【Sources Root】，將文件夾標記為資源文件夾類型;同樣的，選中test目錄下的scala文件夾，單擊右鍵選擇【Mark Directory as】→【Test Sources Root】將文件夾標記為測試資源文件夾類型。其中，資源文件夾中存放項目源碼文件，測試文件夾中存放開發(fā)中測試的源碼文件。

　　2.添加Spark相關依賴、打包插件

　　Maven是一個項目管理工具，雖然我們剛才創(chuàng)建好了項目，但是卻不能識別Spark類，因此，我們需要將Spark相關的依賴添加到Maven項目中。打開pom.xml文件，在該文件中添加的依賴如下所示：

<!--設置依賴版本號-->
     <properties>
      <scala.version>2.11.8</scala.version>
      <hadoop.version>2.7.4</hadoop.version>
      <spark.version>2.3.2</spark.version>
    </properties>
    <dependencies>
      <!--Scala-->
     <dependency>
       <groupId>org.scala-lang</groupId>
       <artifactId>scala-library</artifactId>
       <version>${scala.version}</version>
     </dependency>
     <!--Spark-->
     <dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-core_2.11</artifactId>
       <version>${spark.version}</version>
     </dependency>
     <!--Hadoop-->
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-client</artifactId>
       <version>${hadoop.version}</version>
     </dependency>
   </dependencies>

　　在上述配置參數片段中，標簽用來設置所需依賴的版本號，其中在標簽中添加了Scala、Hadoop和Spark相關的依賴，設置完畢后，相關Jar文件會被自動加載到項目中。

　　3.編寫代碼，查看結果

　　在main目錄下的scala文件夾中，創(chuàng)建WordCount.scala文件用于詞頻統(tǒng)計，代碼如文件1所示。

　　文件1 WordCount.scala

  import org.apache.spark.rdd.RDD
  import org.apache.spark.{SparkConf, SparkContext}
  //編寫單詞計數
  object WordCount {
    def main(args: Array[String]): Unit = {
       //1.創(chuàng)建SparkConf對象,設置appName和Master地址
      val sparkconf = new 
        SparkConf().setAppName("WordCount").setMaster("local[2]")
      //2.創(chuàng)建SparkContext對象，它是所有任務計算的源頭，
     // 它會創(chuàng)建DAGScheduler和TaskScheduler
     val sparkContext = new SparkContext(sparkconf)
     //3.讀取數據文件,RDD可以簡單的理解為是一個集合
     // 集合中存放的元素是String類型
     val data : RDD[String] = 
                  sparkContext.textFile("D:\\word\\words.txt")
     //4.切分每一行，獲取所有的單詞
     val words :RDD[String] = data.flatMap(_.split(" "))
     //5.每個單詞記為1，轉換為（單詞，1）
     val wordAndOne :RDD[(String, Int)] = words.map(x =>(x,1))
     //6.相同單詞匯總，前一個下劃線表示累加數據，后一個下劃線表示新數據
     val result: RDD[(String, Int)] = wordAndOne.reduceByKey(_+_)
     //7.收集打印結果數據
     val finalResult: Array[(String, Int)] = result.collect()
     println(finalResult.toBuffer)
     //8.關閉sparkContext對象
     sparkContext.stop()
   }
 }

　　上述代碼中，第7-11行代碼創(chuàng)建SparkContext對象并通過SparkConf對象設置配置參數，其中Master為本地模式，即可以在本地直接運行;第14-24行代碼中，讀取數據文件，將獲得的數據按照空格切分，將每個單詞記作(單詞，1)，之后若出現相同的單詞就將次數累加，最終打印數據結果;第26行代碼表示關閉SparkContext對象資源。執(zhí)行代碼成功后，在控制臺可以查看輸出結果，如圖2所示。

圖2 IDEA開發(fā)WordCount

　　從圖2可以看出，文本中的單詞已經成功統(tǒng)計了出現的次數。


猜你喜歡：

HBase數據庫物理存儲的存儲方式介紹【大數據文章】

Scala在使用聲明值和變量時的注意事項有哪些？

HBase中常見的Shell命令有哪幾個？具體是怎樣操作的

黑馬程序員大數據培訓課程