1. 程式人生 > >在local模式下的spark程序打包到集群上運行

在local模式下的spark程序打包到集群上運行

inf 分布式 依賴 size wordcount dag 相同 scala 來源

一、前期準備

  前期的環境準備,在Linux系統下要有Hadoop系統,spark偽分布式或者分布式,具體的教程可以查閱我的這兩篇博客:

Hadoop2.0偽分布式平臺環境搭建

Spark2.4.0偽分布式環境搭建

  然後在spark偽分布式的環境下必須出現如下八個節點才算spark環境搭建好。

  技術分享圖片

  然後再在本地windows系統下有一個簡單的詞頻統計程序。

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD

object ScalaSparkDemo {
  def main(args: Array[String]) {
    /**
     * 第一步:創建Spark的配置對象SparkConf,設置Spark程序的運行時的配置信息,
     * 例如說通過setMaster來設置程序要連接的Spark集群的Master的URL,
     * 如果設置為local,則代表Spark程序在本地運行,特別適合於機器配置條件非常差
     * (例如只有1G的內存)的初學者
     */
    val conf = new SparkConf() //創建SparkConf對象,由於全局只有一個SparkConf所以不需要工廠方法
    conf.setAppName("wow,my first spark app") //設置應用程序的名稱,在程序的監控界面可以看得到名稱
    //conf.setMaster("local") //此時程序在本地運行,不需要安裝Spark集群
    /**
     * 第二步:創建SparkContext對象
     * SparkContext是Spark程序所有功能的唯一入口,無論是采用Scala、Java、Python、R等都必須要有一個
     * SparkContext
     * SparkContext核心作用:初始化Spark應用程序運行所需要的核心組件,包括DAGScheduler,TaskScheduler,SchedulerBacked,
     * 同時還會負責Spark程序往Master註冊程序等
     * SparkContext是整個Spark應用程序中最為至關重要的一個對象
     */
    val sc = new SparkContext(conf) //創建SpackContext對象,通過傳入SparkConf實例來定制Spark運行的具體參數的配置信息
    /**
     * 第三步:根據具體的數據來源(HDFS,HBase,Local,FileSystem,DB,S3)通過SparkContext來創建RDD
     * RDD的創建基本有三種方式,(1)根據外部的數據來源(例如HDFS)(2)根據Scala集合(3)由其它的RDD操作
     * 數據會被RDD劃分為成為一系列的Partitions,分配到每個Partition的數據屬於一個Task的處理範疇
     */
    //讀取本地文件並設置為一個Partition
    // val lines = sc.textFile("words.txt", 1) //第一個參數為為本地文件路徑,第二個參數minPartitions為最小並行度,這裏設為1
    sc.setLogLevel("WARN")
    val lines = sc.parallelize(List("pandas","i like pandas"))
    //類型推斷 ,也可以寫下面方式
    // val lines : RDD[String] =sc.textFile("words.txt", 1)
    /**
     * 第四步:對初始的RDD進行Transformation級別的處理,例如map,filter等高階函數
     * 編程。來進行具體的數據計算
     * 第4.1步:將每一行的字符串拆分成單個的單詞
     */
    //對每一行的字符串進行單詞拆分並把所有行的結果通過flat合並成一個大的集合
    val words = lines.flatMap { line => line.split(" ") }
    /**
     * 第4.2步在單詞拆分的基礎上,對每個單詞實例計數為1,也就是word=>(word,1)tuple
     */
    val pairs = words.map { word => (word, 1) }
    /**
     * 第4.3步在每個單詞實例計數為1的基礎之上統計每個單詞在文中出現的總次數
     */
    //對相同的key進行value的累加(包括local和Reduce級別的同時Reduce)
    val wordCounts = pairs.reduceByKey(_ + _)
    //打印結果
    wordCounts.foreach(wordNumberPair => println(wordNumberPair._1 + ":" + wordNumberPair._2))
    //釋放資源
    sc.stop()
  }
}

二、導出jar包

  這裏註意詞頻統計程序的包名為test,類名為ScalaSparkDemo。

  技術分享圖片

   技術分享圖片

  註意這裏勾選要打包所依賴的一些文件。當然可以選擇把整個工程打包。還要註意這裏打包後的文件名為test.jar。

  技術分享圖片

  然後上傳到Ubuntu中,使用這個命令 bin/spark-submit --class test.ScalaSparkDemo --master local /home/xiaow/test.jar 即可運行。/home/xiaow/test.jar:指明此jar包在主節點上的位置。關於打包到集群的詳細命令,可以查閱我的這一篇博客:Spark學習之在集群上運行Spark

  技術分享圖片

    技術分享圖片

  如此,搞定收工!!!

在local模式下的spark程序打包到集群上運行