使用IDEA和SBT構建Spark程序

使用IDEA和SBT構建Spark程序

1. 打開IDEA創建一個Scala項目，選擇sbt

2. 選擇合適的sbt版本和scala版本

3. 創建完成之后會從遠程服務器拉取一些項目的信息，可能比較慢

4. 完成后的項目結構如圖所示

5. 編輯build.sbt文件，導入spark-core依賴

// 可以直接去maven的中央倉庫去找，選擇sbt復制即可
// https://mvnrepository.com/artifact/org.apache.spark/spark-core
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.4.4"

導入過程需要一點時間，因為需要去遠程的中央倉庫去下載，導入完成后，可以在項目左側的External Libraries中看到已經導入的包

6. 建立一個測試類，測試是否配置成功

Spark的WordCount案例

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

object HelloSpark {
    def main(args: Array[String]): Unit = {
        // 創建配置對象
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HelloSpark")
        // 創建上下文對象  SparkContext
        val context: SparkContext = new SparkContext(conf)
        // 從文件中讀取要統計的語句
        val lines: RDD[String] = context.textFile("E:/sparkdata/text.txt")
        // 將從文件讀取到的字符串進行切分
        val words: RDD[String] = lines.flatMap(_.split(" "))
        // 將單詞轉換成為元組
        val tuples: RDD[(String, Int)] = words.map((_, 1))
        // 將元祖進行聚合
        val sumed: RDD[(String, Int)] = tuples.reduceByKey(_ + _)
        // 對聚合后的結果進行排序
        val sorted: RDD[(String, Int)] = sumed.sortBy(_._2, false)
        // 將結果輸出到文件中
        sorted.saveAsTextFile("E:/sparkdata/01")
        // 釋放資源
        context.stop()
    }
}

如果可以運行成功，說明已經配置成功，有的人這里可能會出現運行，這是可以選擇重新配置一下項目的Scala SDK

7. 將WordCount打包成jar包在集群上測試，並將結果保存到集群

如果在集群上運行時，需要將WordCount案例代碼進行修改

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

object HelloSpark {
    def main(args: Array[String]): Unit = {
        // 創建配置對象
        val conf: SparkConf = new SparkConf().setAppName("WordCount")
        // 創建上下文對象  SparkContext
        val context: SparkContext = new SparkContext(conf)
        // 從文件中讀取要統計的語句
        val lines: RDD[String] = context.textFile(args(0))
        // 將從文件讀取到的字符串進行切分
        val words: RDD[String] = lines.flatMap(_.split(" "))
        // 將單詞轉換成為元組
        val tuples: RDD[(String, Int)] = words.map((_, 1))
        // 將元祖進行聚合
        val sumed: RDD[(String, Int)] = tuples.reduceByKey(_ + _)
        // 對聚合后的結果進行排序
        val sorted: RDD[(String, Int)] = sumed.sortBy(_._2, false)
        // 將結果輸出到文件中
        sorted.saveAsTextFile(args(1))
        // 釋放資源
        context.stop()
    }
}

接下來將項目進行打包

• 打開項目結構配置頁面

  

• 添加jar包配置

  

• 如果項目中有多個項目和主類，可以選擇自己要打包的項目和主類

  

• 點擊OK即可，然后去掉額外的lib包依賴，不要將其打包到jar文件中，只保留class編譯文件及META-INF文件夾

  

  

• 編譯生成jar包

  

  

  

• 打包成功后，可以在out中看到jar包

  

• 查看一下jar包內的項目結構

  

• 將打好的jar包上傳到集群，然后在集群上創建一個目錄，存放text.txt

  [hadoop@master ~]$ hdfs dfs -mkdir /sparkdata
  [hadoop@master ~]$ echo "hello world hello spark" >> text.txt
  [hadoop@master ~]$ hdfs dfs -put ./text.txt /sparkdata/
  

• 提交作業，進行測試

  spark-submit \
  --class HelloSpark \
  --executor-memory 512M \
  --total-executor-cores 1 \
  /home/yangqi/sparkdata/SBTSet.jar \
  hdfs://supercluster/sparkdata/text.txt \
  hdfs://supercluster/sparkdata/out/
  

  我這里的集群是高可用，所以使用的supercluster，高可用集群的名稱；如果不是高可用，使用主機名+端口號即可。例如：master:8020

  –class后面一定要添加類全名，由於我的類沒有包，所以直接使用即可