• local 模式

1. 程式碼

package com.imooc.spark.Test

import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SaveMode, SparkSession}

/**
  * 測試sparkContext 案例
  */
object TestOfSparkContext2 {


  def main(args: Array[String]): Unit = {
    //System.setProperty("hadoop.home.dir", "E:\\soft\\winutils\\hadoop-common-2.2.0-bin-master")
    val spark = SparkSession.builder()
      .appName("TestOfSparkContext2")
      .master("local[2]").getOrCreate()

    //反射的方式 RDD=>DF
    // reflection(spark)
    //程式設計的方式 RDD=>DF
    program(spark)
    spark.stop()
  }

  private def reflection(spark: SparkSession) = {
    val rdd = spark.sparkContext
      .textFile("file///home/hadoop/data/test-data-spark/emp-forSpark.txt")
      .map(w => w.split("\t"))

    import spark.implicits._
    val empDF = rdd.map(line =>
      EMP(line(0), line(1), line(2), line(3),
        line(4), line(5), line(6), line(7))).toDF()
    empDF.printSchema()
    empDF.show()
  }

  def program(spark: SparkSession) = {
    val infoRdd = spark.sparkContext
      .textFile("/home/hadoop/data/test-data-spark/emp-forSpark.txt")
      .map(w => w.split("\t")).map(line => Row(line(0), line(1), line(2), line(3), line(4), line(5), line(6), line(7)))


    val filedName = Array(StructField("empNo", StringType, true)
      , StructField("ename", StringType, true), StructField("job", StringType, true)
      , StructField("mgr", StringType, true), StructField("hireDate", StringType, true)
      , StructField("sal", StringType, true), StructField("comm", StringType, true)
      , StructField("deptNo", StringType, true))

    val schema = StructType(filedName)
    val empDf = spark.createDataFrame(infoRdd, schema)
    empDf.printSchema()
     empDf.show(false)

    //註冊臨時表
    empDf.createOrReplaceTempView("emp")
    val sqlDF = spark.sql("select ename,job,comm from emp")
   // sqlDF.printSchema()
    //sqlDF.show()
    /*empDf.write.format("json")
      .mode(SaveMode.Overwrite).save("E:///testData/empTable3")*/
    empDf.coalesce(1).write.format("json").mode(SaveMode.Overwrite)
      .partitionBy("deptno").save("/home/hadoop/app/spark-2.1.1-bin-2.6.0-cdh5.7.0/testData/emp-jsonTable")
  }

  case class EMP(empNo: String, empName: String, job: String, mgr: String, hiredate: String,
                 sal: String, comm: String, deptNo: String)

}

 

2. 打包

3. 提交

在$SPARK_HOME 下執行

./bin/spark-submit \
--class com.imooc.spark.Test.TestOfSparkContext2 \
--master local[2] \
/home/hadoop/data/test-jar/sql-1.0.jar

4. 引數解釋

對empDf.write.format(“json”)
.mode(SaveMode.Overwrite).save(“E:///testData/empTable3”)*/
empDf.coalesce(1).write.format(“json”).mode(SaveMode.Overwrite)
.partitionBy(“deptno”).save("/home/hadoop/app/spark-2.1.1-bin-2.6.0-cdh5.7.0/testData/emp-jsonTable")中的引數解釋

• yarn client模式提交

1. 程式碼

package com.imooc.spark.Test

import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SaveMode, SparkSession}

/**
  * 測試sparkContext 案例
  */
object TestOfSparkContext2OnYarn {


  def main(args: Array[String]): Unit = {
    //System.setProperty("hadoop.home.dir", "E:\\soft\\winutils\\hadoop-common-2.2.0-bin-master")
    /*val spark = SparkSession.builder()
      .appName("TestOfSparkContext2OnYarn")
      .master("local[2]").getOrCreate()*/
    val path = "/user/hadoop/data/test-data-forSpark/emp-forSpark.txt"
    val spark = SparkSession.builder().getOrCreate()


    //反射的方式 RDD=>DF
    // reflection(spark,path)
    //程式設計的方式 RDD=>DF
    program(spark,path)
    spark.stop()
  }

  private def reflection(spark: SparkSession,path:String ) = {
    val rdd = spark.sparkContext
      .textFile(path)
      .map(w => w.split("\t"))

    import spark.implicits._
    val empDF = rdd.map(line =>
      EMP(line(0), line(1), line(2), line(3),
        line(4), line(5), line(6), line(7))).toDF()
    empDF.printSchema()
    empDF.show()
  }

  def program(spark: SparkSession,path:String) = {
    val infoRdd = spark.sparkContext
      .textFile(path)
      .map(w => w.split("\t")).map(line => Row(line(0), line(1), line(2), line(3), line(4), line(5), line(6), line(7)))


    val filedName = Array(StructField("empNo", StringType, true)
      , StructField("ename", StringType, true), StructField("job", StringType, true)
      , StructField("mgr", StringType, true), StructField("hireDate", StringType, true)
      , StructField("sal", StringType, true), StructField("comm", StringType, true)
      , StructField("deptNo", StringType, true))

    val schema = StructType(filedName)
    val empDf = spark.createDataFrame(infoRdd, schema)
    empDf.printSchema()
     empDf.show(false)

    //註冊臨時表
    empDf.createOrReplaceTempView("emp")
    val sqlDF = spark.sql("select ename,job,comm from emp")
   // sqlDF.printSchema()
    //sqlDF.show()
    /*empDf.write.format("json")
      .mode(SaveMode.Overwrite).save("E:///testData/empTable3")*/
    empDf.coalesce(1).write.format("json").mode(SaveMode.Overwrite)
      .partitionBy("deptno").save("hdfs://hadoop001:9000/user/hadoop/emp-spark-test/emp-jsonOnYarnTable")
  }

  case class EMP(empNo: String, empName: String, job: String, mgr: String, hiredate: String,
                 sal: String, comm: String, deptNo: String)

}

2. 提交

./bin/spark-submit \
--class com.imooc.spark.Test.TestOfSparkContext2OnYarn \
--master yarn \
/home/hadoop/data/jar-test/sql-3.0-onYarn.jar

./bin/spark-submit \
--class com.imooc.spark.Test.TestOfSparkContext2OnYarn \
--master yarn \
--deploy-mode client \
/home/hadoop/data/jar-test/sql-3.0-onYarn.jar

注意:

spark-submit 提交jar包到yarn上的時候,資料輸入路徑,資料輸出路徑都必須是HDFS的路徑,否則報錯 :Input path does not exist

3. 檢視

• 日誌直接在控制檯輸出
  

4. 結果

[[email protected] ~]$ hadoop fs -ls /user/hadoop/emp-spark-test/emp-jsonOnYarnTable
18/11/20 17:42:27 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 5 items
-rw-r--r--   1 hadoop supergroup          0 2018-11-20 17:33 /user/hadoop/emp-spark-test/emp-jsonOnYarnTable/_SUCCESS
drwxr-xr-x   - hadoop supergroup          0 2018-11-20 17:33 /user/hadoop/emp-spark-test/emp-jsonOnYarnTable/deptno=10
drwxr-xr-x   - hadoop supergroup          0 2018-11-20 17:33 /user/hadoop/emp-spark-test/emp-jsonOnYarnTable/deptno=20
drwxr-xr-x   - hadoop supergroup          0 2018-11-20 17:33 /user/hadoop/emp-spark-test/emp-jsonOnYarnTable/deptno=30
drwxr-xr-x   - hadoop supergroup          0 2018-11-20 17:33 /user/hadoop/emp-spark-test/emp-jsonOnYarnTable/deptno=44
[[email protected] ~]$ 

注意

• 本地提交:輸入輸出路檔案在本地寫法:file:///
• 輸入輸出檔案在HDFS寫法:hdfs://ip:port/
• 其中提交模式是yarn模式的時候,輸入輸出只能是在hdfs上操作

• yarn cluster 模式

1. 提交命令

sql-1.0-yarnCluster.jar
./bin/spark-submit \
--class com.imooc.spark.Test.TestOfSparkContext2OnYarn \
--master yarn-cluster \
/home/hadoop/data/jar-test/sql-3.0-onYarn.jar

2. 檢視

• 檢視輸出日誌需要在 yarn的日誌上檢視
  
• Spark on YARN 之 client 與cluster 模式區別

• standalone 模式

1. 程式碼

和local模式一樣和local模式一樣

2. 提交

./bin/spark-submit
–class com.imooc.spark.Test.TestOfSparkContext2
–master spark://192.168.52.130:7077
–executor-memory 4G
–total-executor-cores 6
/home/hadoop/data/jar-test/sql-1.0-yarnCluster.jar

• 三種模式比較

• 提交引數詳解