更新時(shí)間:2020-12-17 來(lái)源:黑馬程序員 瀏覽量:
Spark作業(yè)與MapReduce作業(yè)同樣可以先在本地開(kāi)發(fā)測(cè)試,本地執(zhí)行模式與集群提交模式,代碼的業(yè)務(wù)功能相同,因此本書大多數(shù)采用本地開(kāi)發(fā)模式。下面講解使用IDEA工具開(kāi)發(fā)WordCount單詞計(jì)數(shù)程序的相關(guān)步驟。
1.創(chuàng)建Maven項(xiàng)目,新建資源文件夾
創(chuàng)建一個(gè)Maven工程項(xiàng)目,名為“spark_chapter02”。項(xiàng)目創(chuàng)建好后,在main和test目錄下分別創(chuàng)建一個(gè)名稱為scala的文件夾,創(chuàng)建好的目錄結(jié)構(gòu)如圖1所示。
圖1 Spark_chapter02項(xiàng)目目錄結(jié)構(gòu)
在圖1中,選中main目錄下的scala文件夾,單擊右鍵選擇【Mark Directory as】→【Sources Root】,將文件夾標(biāo)記為資源文件夾類型;同樣的,選中test目錄下的scala文件夾,單擊右鍵選擇【Mark Directory as】→【Test Sources Root】將文件夾標(biāo)記為測(cè)試資源文件夾類型。其中,資源文件夾中存放項(xiàng)目源碼文件,測(cè)試文件夾中存放開(kāi)發(fā)中測(cè)試的源碼文件。
2.添加Spark相關(guān)依賴、打包插件
Maven是一個(gè)項(xiàng)目管理工具,雖然我們剛才創(chuàng)建好了項(xiàng)目,但是卻不能識(shí)別Spark類,因此,我們需要將Spark相關(guān)的依賴添加到Maven項(xiàng)目中。打開(kāi)pom.xml文件,在該文件中添加的依賴如下所示:
<!--設(shè)置依賴版本號(hào)--> <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>
在上述配置參數(shù)片段中,標(biāo)簽用來(lái)設(shè)置所需依賴的版本號(hào),其中在標(biāo)簽中添加了Scala、Hadoop和Spark相關(guān)的依賴,設(shè)置完畢后,相關(guān)Jar文件會(huì)被自動(dòng)加載到項(xiàng)目中。
3.編寫代碼,查看結(jié)果
在main目錄下的scala文件夾中,創(chuàng)建WordCount.scala文件用于詞頻統(tǒng)計(jì),代碼如文件1所示。
文件1 WordCount.scala
import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} //編寫單詞計(jì)數(shù) object WordCount { def main(args: Array[String]): Unit = { //1.創(chuàng)建SparkConf對(duì)象,設(shè)置appName和Master地址 val sparkconf = new SparkConf().setAppName("WordCount").setMaster("local[2]") //2.創(chuàng)建SparkContext對(duì)象,它是所有任務(wù)計(jì)算的源頭, // 它會(huì)創(chuàng)建DAGScheduler和TaskScheduler val sparkContext = new SparkContext(sparkconf) //3.讀取數(shù)據(jù)文件,RDD可以簡(jiǎn)單的理解為是一個(gè)集合 // 集合中存放的元素是String類型 val data : RDD[String] = sparkContext.textFile("D:\\word\\words.txt") //4.切分每一行,獲取所有的單詞 val words :RDD[String] = data.flatMap(_.split(" ")) //5.每個(gè)單詞記為1,轉(zhuǎn)換為(單詞,1) val wordAndOne :RDD[(String, Int)] = words.map(x =>(x,1)) //6.相同單詞匯總,前一個(gè)下劃線表示累加數(shù)據(jù),后一個(gè)下劃線表示新數(shù)據(jù) val result: RDD[(String, Int)] = wordAndOne.reduceByKey(_+_) //7.收集打印結(jié)果數(shù)據(jù) val finalResult: Array[(String, Int)] = result.collect() println(finalResult.toBuffer) //8.關(guān)閉sparkContext對(duì)象 sparkContext.stop() } }
上述代碼中,第7-11行代碼創(chuàng)建SparkContext對(duì)象并通過(guò)SparkConf對(duì)象設(shè)置配置參數(shù),其中Master為本地模式,即可以在本地直接運(yùn)行;第14-24行代碼中,讀取數(shù)據(jù)文件,將獲得的數(shù)據(jù)按照空格切分,將每個(gè)單詞記作(單詞,1),之后若出現(xiàn)相同的單詞就將次數(shù)累加,最終打印數(shù)據(jù)結(jié)果;第26行代碼表示關(guān)閉SparkContext對(duì)象資源。執(zhí)行代碼成功后,在控制臺(tái)可以查看輸出結(jié)果,如圖2所示。
圖2 IDEA開(kāi)發(fā)WordCount
從圖2可以看出,文本中的單詞已經(jīng)成功統(tǒng)計(jì)了出現(xiàn)的次數(shù)。
猜你喜歡:
HBase數(shù)據(jù)庫(kù)物理存儲(chǔ)的存儲(chǔ)方式介紹【大數(shù)據(jù)文章】
Scala在使用聲明值和變量時(shí)的注意事項(xiàng)有哪些?
Scala在使用聲明值和變量時(shí)的注意事項(xiàng)有哪些?
2020-12-17詳解MapReduce編程的Map模型和Reduce模型
2020-12-17HBase數(shù)據(jù)庫(kù)物理存儲(chǔ)的存儲(chǔ)方式介紹【大數(shù)據(jù)文章】
2020-12-17HDFS分布式文件系統(tǒng)的優(yōu)點(diǎn)缺點(diǎn)有哪些?
2020-12-16不同系統(tǒng)如何加載數(shù)據(jù)創(chuàng)建RDD?
2020-12-07什么是數(shù)據(jù)倉(cāng)庫(kù)?Hive數(shù)據(jù)倉(cāng)庫(kù)特點(diǎn)簡(jiǎn)介
2020-12-07