object Operate_Text {
    def main(args: Array[String]): Unit = {
        //1.創(chuàng)建SparkConf并設(shè)置App名稱
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")
        //2.創(chuàng)建SparkContext，該對(duì)象是提交Spark App的入口
        val sc: SparkContext = new SparkContext(conf)
        //3.1 讀取輸入文件
        val inputRDD: RDD[String] = sc.textFile("input/demo.txt")
        //3.2 保存數(shù)據(jù)
        inputRDD.saveAsTextFile("textFile")
        //4.關(guān)閉連接
        sc.stop()
    }
}

object Operate_Sequence {
    def main(args: Array[String]): Unit = {
        //1.創(chuàng)建SparkConf并設(shè)置App名稱
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")
        //2.創(chuàng)建SparkContext，該對(duì)象是提交Spark App的入口
        val sc: SparkContext = new SparkContext(conf)
        //3.1 創(chuàng)建rdd
        val dataRDD: RDD[(Int, Int)] = sc.makeRDD(Array((1,2,3),(4,5,6),(7,8,9)))
        //3.2 保存數(shù)據(jù)為SequenceFile
        dataRDD.saveAsSequenceFile("seqFile")
        //3.3 讀取SequenceFile文件
        sc.sequenceFile[Int,Int]("seqFile").collect().foreach(println)
        //4.關(guān)閉連接
        sc.stop()
    }
}

object Operate_Object {
    def main(args: Array[String]): Unit = {
        //1.創(chuàng)建SparkConf并設(shè)置App名稱
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")
        //2.創(chuàng)建SparkContext，該對(duì)象是提交Spark App的入口
        val sc: SparkContext = new SparkContext(conf)
        //3.1 創(chuàng)建RDD
        val dataRDD: RDD[Int] = sc.makeRDD(Array(1,2,3,4,5,6),2)
        //3.2 保存數(shù)據(jù)
        dataRDD.saveAsObjectFile("objFile")
        //3.3 讀取數(shù)據(jù)
        sc.objectFile[Int]("objFile").collect().foreach(println)
        //4.關(guān)閉連接
        sc.stop()
    }
}

object accumulator_system {
package com.atguigu.cache
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}
object accumulator_system {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val dataRDD: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)))
    //需求:統(tǒng)計(jì)a出現(xiàn)的所有次數(shù) ("a",10)
    //普通算子實(shí)現(xiàn) reduceByKey 代碼會(huì)走shuffle 效率低
    val rdd: RDD[(String, Int)] = dataRDD.reduceByKey(_ + _)
    //累加器實(shí)現(xiàn)
    //1 聲明累加器
    val accSum: LongAccumulator = sc.longAccumulator("sum")
    dataRDD.foreach{
      case (a,count) => {
        //2 使用累加器累加  累加器.add()
        accSum.add(count)
        // 4 不在executor端獲取累加器的值,因?yàn)榈玫降闹挡粶?zhǔn)確，所以累加器叫分布式共享只寫(xiě)變量
        //println("sum = " + accSum.value)
      }
    }
    //3 獲取累加器的值 累加器.value
    println(("a",accSum.value))
    sc.stop()
  }
}