package cn.spark.sparkbase

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

/**
 * DESC:
 * 1-首先创建SparkContent上下文环境
 * 2-从外部文件数据源读取数据
 * 3-执行flatmap执行扁平化操作
 * 4-执行map转化操作，(word,1)
 * 5-reduceByKey(预聚合)
 * 6-输出到文件系统或打印即可
 */
object _01SparkWordCount {
  def main(args: Array[String]): Unit = {
    //* 1-首先创建SparkContext上下文环境
    val conf: SparkConf = new SparkConf().setAppName("_01SparkWordCount").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)
    //sc.setLogLevel("WARN")
    //* 2-从外部文件数据源读取数据
    val fileRDD: RDD[String] = sc.textFile("data/baseinput/words.txt")
    //println(s"count value is:${fileRDD.count()}")//count value is:2
    //* 3-执行flatmap执行扁平化操作
    val valueRDD: RDD[String] = fileRDD.flatMap(x => x.split("\\s+"))
    //* 4-执行map转化操作，(word,1)
    val mapRDD: RDD[(String, Int)] = valueRDD.map(x => (x, 1))
    //* 5-reduceByKey(预聚合)
    val resultRDD: RDD[(String, Int)] = mapRDD.reduceByKey((a, b) => a + b)
    //* 6-输出到文件系统或打印即可
    resultRDD.foreach(x=>println(x)) //这里是直接使用foreach对RDD进行打印
    println("=============================")
    resultRDD.collect().foreach(println(_))//能否转换为集合在打印
    println("=============================")
    resultRDD.saveAsTextFile("data/baseoutput/output-1")
    sc.stop()
  }
}

package cn.spark.sparkbase.rddopration

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

/**
 * DESC:
 */
object _02SparkRDDFirst {
  def main(args: Array[String]): Unit = {
    //1-准备环境
    val conf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
      .set("spark.default.parallelism", "4")
    val sc = new SparkContext(conf)
    //2-准备读取外部文件系统
    val fileRDD: RDD[String] = sc.textFile("data/baseinput/words.txt")
    //sc.parallelize(1 to 10)
    //3-查看分区个数
    println("partitons length:" + fileRDD.getNumPartitions) //2
    println(s"partiiton length is:${fileRDD.partitions.length}") //2
    //读取一个文件夹下的多个文件
    //textFile在读取小文件的时候，会参考小文件的个数，文件个数越多，分区个数越多
    val filesRDD1: RDD[String] = sc.textFile("data/baseinput/ratings100/")
    println("partitons length:" + filesRDD1.getNumPartitions) //100
    //Spark读取小文件的方法
    val filesRDD2: RDD[(String, String)] = sc.wholeTextFiles("data/baseinput/ratings100/")
    //println("partitons length:" + filesRDD2.getNumPartitions) //2
    filesRDD2.take(3).foreach(println(_))
    //23106,5214,3.0,1043445719

    sc.stop()
  }
}

package cn.spark.sparkbase.base

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

import scala.collection.mutable.ArrayBuffer

/**
 * DESC:
 */
object _01groupByKeyOperation {
  def main(args: Array[String]): Unit = {
    //* 1-首先创建SparkContext上下文环境
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)
    //* 2创建数据，生成RDD
    val filerdd: RDD[String] = sc.makeRDD(Array("hello you", "hello me hello she", "hello spark"))
    val flatmapRDD: RDD[String] = filerdd.flatMap(_.split("\\s+"))
    val mapRDD: RDD[(String, Int)] = flatmapRDD.map(x => (x, 1))
    //* 3使用groupBy完成分组
    val groupRDD1: RDD[(String, Iterable[Int])] = mapRDD.groupByKey()
    groupRDD1.collect().foreach(println(_))
    //(me,CompactBuffer(1))
    //(spark,CompactBuffer(1))
    //(you,CompactBuffer(1))
    //(she,CompactBuffer(1))
    //(hello,CompactBuffer(1, 1, 1, 1))
    //CompactBuffer是spark模仿arraybuffer构建缓冲容器
    //换一种方法模拟gropuByKey实现相同key的value的分组，使用源码的方法
    //1-首先CompactBuffer存放分组的结果可以防止的ArrayBuffer
    //val createCombiner = (v: V) => CompactBuffer(v)
    //val mergeValue = (buf: CompactBuffer[V], v: V) => buf += v
    //val mergeCombiners = (c1: CompactBuffer[V], c2: CompactBuffer[V]) => c1 ++= c2
    /*
    key value 存放value的arraybuffer
     */
    val createCombiner = (v: Int) => ArrayBuffer(v)
    val mergeValue = (buf: ArrayBuffer[Int], v: Int) => buf += v
    val mergeCombiners = (c1: ArrayBuffer[Int], c2: ArrayBuffer[Int]) => c1 ++= c2
    val valueRDD: ShuffledRDD[String, Int, ArrayBuffer[Int]] = new ShuffledRDD[String, Int, ArrayBuffer[Int]](mapRDD, new org.apache.spark.HashPartitioner(4))
      .setAggregator(new Aggregator(
        createCombiner,
        mergeValue,
        mergeCombiners))
      .setMapSideCombine(false)
    println("source code create groupBykey")
    valueRDD.foreach(println(_))
    sc.stop()
  }
}

package cn.spark.sparkbase.base

import org.apache.spark.{SparkConf, SparkContext, TaskContext}
import org.junit.Test

/**
 * DESC:
 */
class _02RDDTest {

  private val sc = new SparkContext(new SparkConf().setAppName("_02RDDTest").setMaster("local[*]"))

  @Test
  def test01(): Unit = {
    sc.parallelize(Seq(1, 2, 3, 4, 5))
      .map(x => x * 2)
      .foreach(println(_))

    sc.parallelize(Seq(1, 2, 3, 4, 5))
      .filter(x => x > 3)
      .foreach(println(_))

  }

  @Test
  def test02: Unit = {
    sc.parallelize(Array(1, 2, 3, 4, 5, 6), 2)
      .map(x => (x * 2))
      .foreach(println(_))
    //如果直接使用foreach是无法作为iteratale返回
    sc.parallelize(Array(1, 2, 3, 4, 5, 6), 2)
      //f: Iterator[T] => Iterator[U],
      .mapPartitions(iter => {
        iter.foreach(println(_))
        iter
      }).collect()
    //执行每个分区的元素乘以2
    sc.parallelize(Array(1, 2, 3, 4, 5, 6), 2)
      //f: Iterator[T] => Iterator[U],
      .mapPartitions(iter => {
        val iterator: Iterator[Int] = iter.map(item => item * 2)
        iterator
      }).collect().foreach(println(_))

    println("上面的等价写法")
    sc.parallelize(Array(1, 2, 3, 4, 5, 6), 2)
      //f: Iterator[T] => Iterator[U],
      .mapPartitions(iter => {
        iter.map(item => item * 2)
      }).collect().foreach(println(_))
  }

  @Test
  def test03: Unit = {
    sc.parallelize(Array(1, 2, 3, 4, 5, 6), 2)
      // f: (Int几号分区, Iterator[T]) => Iterator[U]
      .mapPartitionsWithIndex((index, iter) => {
        println("index:" + index)
        iter.map(_ * 2)
      }).collect().foreach(println(_))
    println("改进的方法")
    sc.parallelize(Array(1, 2, 3, 4, 5, 6), 2)
      // f: (Int几号分区, Iterator[T]) => Iterator[U]
      .mapPartitionsWithIndex((index, iter) => {
        iter.map(x => "index is:" + index + "\tvalue is:" + x * 2)
      }).collect().foreach(println(_))
    println("想用mapPartition的方法实现分区有哪些元素")
    sc.parallelize(Array(1, 2, 3, 4, 5, 6), 2)
      .mapPartitions(iter => {
        println("partitionID:", TaskContext.getPartitionId())
        iter.map(x => x * 2)
      }).collect().foreach(println(_))

    println("想用mapPartition的方法实现分区有哪些元素,改进方法")
    sc.parallelize(Array(1, 2, 3, 4, 5, 6), 2)
      .mapPartitions(iter => {
        iter.map(x => "partitionID:" + TaskContext.getPartitionId() + "\tvalue is:" + x * 2)
      }).collect().foreach(println(_))
    //想用mapPartition的方法实现分区有哪些元素,改进方法
    //partitionID:0  value is:2
    //partitionID:0  value is:4
    //partitionID:0  value is:6
    //partitionID:1  value is:8
    //partitionID:1  value is:10
    //partitionID:1  value is:12
  }

}

@Test
def test01(): Unit = {
  val value: RDD[(String, Int)] = sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
  value
    .groupByKey()
    .collect()
    .foreach(println(_))
  //(a,CompactBuffer(1, 1))
  //(b,CompactBuffer(1))
  value
    .groupByKey()
    .map(x => (x._1, x._2.sum))
    .collect()
    .foreach(println(_))
}

@Test
def test02: Unit = {
  val rdd: RDD[(String, Double)] = sc.parallelize(Seq(
    ("zhangsan", 99.0),
    ("zhangsan", 96.0),
    ("lisi", 97.0),
    ("lisi", 98.0),
    ("zhangsan", 97.0))
  )
  // createCombiner: Value 元素=> C 对元素的操作,
  // mergeValue: (C, Value) => C,需要将上一个初始值和新的value进行合并
  // mergeCombiners: (C, C) => C,
  val createCombiner = (curr: Double) => (curr, 1)
  val mergeValue = (curr: (Double, Int), nextValue: Double) => (curr._1 + nextValue, curr._2 + 1)
  val mergeCombiners = (curr1: (Double, Int), agg1: (Double, Int)) => (curr1._1 + agg1._1, curr1._2 + agg1._2)
  val valueRDD: RDD[(String, (Double, Int))] = rdd.combineByKey(createCombiner, mergeValue, mergeCombiners)

  valueRDD.foreach(println(_))
  // x._1      x._2._1
  //(zhangsan,(292.0,3))
  //(lisi,(195.0,2))
  valueRDD.map(x => (x._1, x._2._1 / x._2._2)).foreach(println(_))//平均值
  //(zhangsan,(292.0,3))
  //(zhangsan,97.33333333333333)
  //(lisi,97.5)
}

@Test
def test03: Unit = {
  //
  sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
    .aggregateByKey(0)(_ + _, _ + _)
    .collect()
    .foreach(println(_))

  sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
    .foldByKey(0)(_ + _)
    .collect()
    .foreach(println(_))

  sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
    .foldByKey(0)((a, b) => a + b)
    .collect()
    .foreach(println(_))
}

@Test
def test04(): Unit ={
  val rdd = sc.parallelize(List(("a", 1), ("a", 3), ("b", 2)))
  rdd.collectAsMap().foreach(println(_))
}

@Test
def test04(): Unit ={
  val rdd = sc.parallelize(List(("a", 1), ("a", 3), ("b", 2)))
  rdd.collectAsMap().foreach(println(_))
  //Randomly splits this RDD with the provided weights.
  //weights: Array[Double],
  //seed: Long = Utils.random.nextLong
  val array: Array[RDD[(String, Int)]] = rdd.randomSplit(Array(0.6, 0.4), 123L)
  val traingSet: RDD[(String, Int)] = array(0)
  val testSet: RDD[(String, Int)] = array(1)
  println("=======================")
  traingSet.collect().foreach(println(_))
}

@Test
def testJoin(): Unit ={
  // 模拟数据集
  val empRDD: RDD[(Int, String)] = sc.parallelize(
    Seq((1001, "zhangsan"), (1002, "lisi"), (1003, "wangwu"), (1004, "zhangliu"))
  )
  val deptRDD: RDD[(Int, String)] = sc.parallelize(
    Seq((1001, "sales"), (1002, "tech"))
  )
  empRDD.join(deptRDD).foreach(println(_))
  println("============")
  empRDD.leftOuterJoin(deptRDD).foreach(println(_))
}

package cn.spark.sparkbase.pro

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

/**
 * DESC:
 * 实现PV-PageView：
 * 1-首先读取数据
 * 2-通过map转换函数对于每一行数据都统计为一个PV
 * 3-需要通过reduceByKey累加计算(AggerateByKey，FoldByKey)
 * 4-实现PV排序操作
 * 5-打印输出或保存起来
 * 实现UV-UserView
 */
object _01SparkPvUvTopK {
  def main(args: Array[String]): Unit = {
    // 实现PV-PageView：
    val sc: SparkContext = {
      val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")
      val sc = new SparkContext(conf)
      sc
    }
    // 1-首先读取数据
    val fileRDD: RDD[String] = sc.textFile("data/baseinput/access.log")
    //fileRDD.take(5).foreach(println(_))
    //194.237.142.21 - - [18/Sep/2013:06:49:18 +0000] "GET /wp-content/uploads/2013/07/rstudio-git3.png HTTP/1.1" 304 0 "-" "Mozilla/4.0 (compatible;)"
    // 2-通过map转换函数对于每一行数据都统计为一个PV
    val mapRDD: RDD[(String, Int)] = fileRDD.map(x => ("PV", 1))
    //mapRDD.take(3).foreach(println(_))
    // 3-需要通过reduceByKey累加计算(AggerateByKey，FoldByKey)
    val resuleRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)
    // 5-打印输出或保存起来
    resuleRDD.foreach(println(_)) //(PV,14619)
    // 实现UV-UserView
    // 1-读取数据 -已经读取
    // 2-通过map转换函数将IP地址选择出来
    val ipValue: RDD[String] = fileRDD.map(x => x.split("\\s+")).map(x => x(0))
    // 3-使用distinct去重，并实现统计
    val uvValue: RDD[(String, Int)] = ipValue.distinct().map(x => ("UV", 1))
    // 4-使用reduceByKey实现相同key的value的统计
    val uvResult: RDD[(String, Int)] = uvValue.reduceByKey(_ + _)
    uvResult.foreach(println(_)) //(UV,1050)
    // 5-实现UV的排序输出
    //实现TopK-排名前几位
    //1-读取数据--已经读取
    //2-通过map选择出用户点击的URL或APP等，X(10)
    val value1RDD: RDD[(String, Int)] = fileRDD.map(_.split("\\s+")).filter(x => x.length > 10).map(x => (x(10), 1))
    //value1RDD.take(20).foreach(println(_))
    //3-进一步实现过滤统计
    val result1RDD: RDD[(String, Int)] = value1RDD.reduceByKey(_ + _).sortBy(x => x._2, false).filter(x => x._1 != "\"-\"")
    //如果直接执行sortBy或sortBykey全局排序的，打印在客户端是随机的分区，这时候可以增加collect收集到driver端，但是数据量大不建议，collect影响性能。
    result1RDD.take(5).foreach(println(_))
    //("http://blog.fens.me/category/hadoop-action/",547)
    //("http://blog.fens.me/",377)
    //("http://blog.fens.me/wp-admin/post.php?post=2445&action=edit&message=10",360)
    //("http://blog.fens.me/r-json-rjson/",274)
    //("http://blog.fens.me/angularjs-webstorm-ide/",271)
    //4-实现TopK输出
    sc.stop()
  }
}

package cn.spark.sparkbase.pro

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

/**
 * DESC:
 * 1-创建SparkContext环境
 * 2-读取两个文件，ip和用户地址
 * 3-将用户IP对比IP信息段
 * 3-1首先将用户ip转化为long类型
 * 3-2将long类型ip放在ip地址信息段中进行查询（二分查找法）
 * 3-3根据查找的索引下标，得到经度和维度
 * 4-根据经度和维度统计区域热度
 */
object _05IPCheck {
/
  def binarySearch(ipLong: Long, boradcastValue: Array[(String, String, String, String)]): Int = {
    var start = 0
    var end = boradcastValue.length - 1

    while (start <= end) {
      var middle = (start + end) / 2
      //查找出结果--中间位置的startip
      if (ipLong >= boradcastValue(middle)._1.toLong && ipLong <= boradcastValue(middle)._2.toLong) {
        return middle
      }
      //在右边查找
      if (ipLong > boradcastValue(middle)._2.toLong) {
        start = middle + 1
      }
      //在左边查找
      if (ipLong < boradcastValue(middle)._1.toLong) {
        end = middle - 1
      }
    }
    -1
  }

  def ipToLong(ip: String): Long = {
    //注意:IP个原始面貌:
    //10111111.10111010.11110000.11110000
    val ipArr: Array[Int] = ip.split("[.]").map(s => Integer.parseInt(s))
    var ipnum = 0L
    for (i <- ipArr) {
      //<<表示位运算左移 ,0L左移之后还是0L,二进制形式:00000000.00000000.00000000.00000000
      //其他数,左移之后,后面补0
      //|表示位运算或,或的特点是,与0进行或,返回本身
      //第一次:
      //00000000.00000000.00000000.10111111
      //00000000.00000000.00000000.00000000
      //00000000.00000000.00000000.10111111
      //第二次:
      //00000000.00000000.00000000.10111010
      //00000000.00000000.10111111.00000000
      //00000000.00000000.10111111.10111010
      //第三次:
      //00000000.00000000.00000000.11110000
      //00000000.10111111.10111010.00000000
      //00000000.10111111.10111010.11110000
      //第四次:
      //0000
        0.00000000.00000000.11110000
      //10111111.10111010.11110000.00000000
      //10111111.10111010.11110000.11110000
      ipnum = i | (ipnum << 8)
    }
    ipnum
  }

  def main(args: Array[String]): Unit = {
    //1-创建SparkContext环境
    val sc: SparkContext = {
      val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")
      val sc = new SparkContext(conf)
      sc
    }
    //2-读取两个文件，ip和用户地址
    val userIPRDD: RDD[String] = sc.textFile("data/baseinput/ip/20190121000132.394251.http.format")
    val userSplitRDD: RDD[String] = userIPRDD.map(_.split("\\|")).map(x => x(1))
    //userSplitRDD.take(5).foreach(println(_))//125.213.100.123

    val ipRangeRDD: RDD[String] = sc.textFile("data/baseinput/ip/ip.txt")
    //223.247.0.0|223.247.7.255|3757506560|3757508607|亚洲|中国|安徽|池州||电信|341700|China|CN|117.489157|30.656037
    //开始ip的起始long类型地址，结束ip的long类型地址，经度，维度
    val ipRDD: RDD[(String, String, String, String)] = ipRangeRDD.map(_.split("\\|")).map(x => (x(2), x(3), x(x.length - 2), x(x.length - 1)))
    //ipRDD.take(5).foreach(println(_))//(16777472,16778239,119.306239,26.075302)
    val broadcastIpValue: Broadcast[Array[(String, String, String, String)]] = sc.broadcast(ipRDD.collect())
    //broadcastIpValue.value//直接引用他的值
    //3-将用户IP对比IP信息段
    val resultRDD: RDD[((String, String), Int)] = userSplitRDD.mapPartitions(iter => {
      val boradcastValue: Array[(String, String, String, String)] = broadcastIpValue.value
      iter.map(ip => {
        //* 3-1首先将用户ip转化为long类型
        val ipLong: Long = ipToLong(ip)
        //* 3-2将long类型ip放在ip地址信息段中进行查询（二分查找法）
        val index: Int = binarySearch(ipLong, boradcastValue)
        //* 3-3根据查找的索引下标，得到经度和维度
        ((boradcastValue(index)._3, boradcastValue(index)._4), 1)
      })
    }) //end mapParttion
    //4-根据经度和维度统计区域热度
    resultRDD.reduceByKey(_ + _).sortBy(_._2, false).take(5).foreach(println(_))
    //((108.948024,34.263161),1824)
    //((116.405285,39.904989),1535)
    //((106.504962,29.533155),400)
    //((114.502461,38.045474),383)
    //((106.57434,29.60658),177)
    sc.stop()
  }
}

package cn.spark.sparkbase.pro

import java.util

import com.hankcs.hanlp.HanLP
import com.hankcs.hanlp.seg.common.Term
import com.hankcs.hanlp.tokenizer.StandardTokenizer

/**
 * DESC:
 */
import scala.collection.JavaConverters._

object _02HanLp {
  def main(args: Array[String]): Unit = {

    val terms: util.List[Term] = HanLP.segment("我是黑马程序员的一匹黑马")
    println(terms) //[我/r, 是/v, 黑马/n, 程序员/n, 的/uj, 一/m, 匹/q, 黑马/n]
    println(terms.asScala.map(_.word.trim)) //ArrayBuffer(我, 是, 黑马, 程序员, 的, 一, 匹, 黑马)

    val terms1: util.List[Term] = StandardTokenizer.segment("我是黑马程序员的一匹黑马")
    println(terms1.asScala.map(_.word.trim))

    //00:00:00 2982199073774412 [360安全卫士]  8 3  download.it.com.cn/softweb/software/firewall/antivirus/20067/17938.html
    val arr: Array[String] = """00:00:00 2982199073774412 [360安全卫士]  8 3  download.it.com.cn/softweb/software/firewall/antivirus/20067/17938.html""".split("\\s+")
    println(arr(2).replaceAll("\\[|\\]", ""))//360安全卫士
  }
}

package cn.spark.sparkbase.pro

import java.util

import com.hankcs.hanlp.HanLP
import com.hankcs.hanlp.seg.common.Term
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.JavaConverters._
import scala.collection.mutable

/**
 * DESC:
 * 首先完成搜狗词库的数据读取
 * 1-创建SparkContext
 * 2-读取数据
 */
object _04SougouCount {
  def main(args: Array[String]): Unit = {
    //1-创建SparkContext
    val sc: SparkContext = {
      val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")
      val sc = new SparkContext(conf)
      sc
    }
    //2-读取数据
    val sougouRDD: RDD[String] = sc.textFile("data/baseinput/sougu/SogouQ.reduced")
    println(s"sougouRDD count value is:${sougouRDD.count()}") //1724264
    //3-引入样例类，可以基于样例类更好解析各个搜狗的字段
    val recordRDD: RDD[SogouRecord] = sougouRDD
      //这里的过滤需要处理，需要对有缺失字段的需要使用length判断长度
      .filter(line => line != null && line.trim.split("\\s+").length == 6)
      .mapPartitions(iter => {
        iter.map(record => {
          val arr: Array[String] = record.split("\\s+")
          SogouRecord(arr(0), arr(1), arr(2).replaceAll("\\[|\\]", ""), arr(3).toInt, arr(4).toInt, arr(5))
        })
      })
      //打印信息
    recordRDD.take(5).foreach(println(_))
    //SogouRecord(00:00:00,2982199073774412,360安全卫士,8,3,download.it.com.cn/softweb/software/firewall/antivirus/20067/17938.html)
    //SogouRecord(00:00:00,07594220010824798,哄抢救灾物资,1,1,news.21cn.com/social/daqian/2008/05/29/4777194_1.shtml)
    //3-搜索关键词统计
    val valueRDD: RDD[String] = recordRDD.mapPartitions(iter => {
      iter.flatMap(record => {
        val terms: util.List[Term] = HanLP.segment(record.queryWords)
        terms.asScala.map(_.word.trim)
      })
    }) //end  mappartition
    //valueRDD.take(5).foreach(println(_))
    val keyWordCount: RDD[(String, Int)] = valueRDD.map(x => (x, 1)).reduceByKey(_ + _).sortBy(_._2, false)
    println("========搜索关键词统计============")
    keyWordCount.take(5).foreach(println(_))
    //(+,193939)
    //(的,93246)
    //(.,90985)
    //(地震,88451)
    //(救灾,69662)
    

  sc.stop()
  }
}

package cn.spark.sparkbase.base

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

/**
 * DESC:
 */
object _04broadcast {
  def main(args: Array[String]): Unit = {

    //申请资源
    val sc:SparkContext={
      val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")
      val sc = new SparkContext(conf)
      sc
    }
    //创建RDD
    val kvFruit: RDD[(Int, String)] = sc.parallelize(List((1,"apple"),(2,"orange"),(3,"banana"),(4,"grape")))
    val fruitMap: collection.Map[Int, String] =kvFruit.collectAsMap
    //fruitMap.foreach(println(_))
    //需求：根据水果的编号查找水果的名称
    val fruitsIds: RDD[Int] = sc.parallelize(Array(2, 4, 1, 3))
    fruitsIds.map(x=>fruitMap(x)).collect().foreach(println(_))
    //改进：如果水果很多，那么每个水果都需要拉取fruitMap变量进行对比得到水果名称
    val broadMap: Broadcast[collection.Map[Int, String]] = sc.broadcast(fruitMap)
    fruitsIds.map(x=>broadMap.value(x)).collect().foreach(println(_))
    //orange
    //grape
    //apple
    //banana
    sc.stop()
  }
}

package cn.spark.sparkbase.base

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

/**
 * DESC:
 */
object _05accumulate {
  def main(args: Array[String]): Unit = {
    //申请资源
    val sc: SparkContext = {
      val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")
      val sc = new SparkContext(conf)
      sc
    }
    //scala的加法
    var counter1 = 0
    val seq = Seq(1, 2, 3)
    seq.map(x => counter1 += x)
    println("counter result is:" + counter1)
    //rdd的加法-0--为什么会出现现象？因为变量在driver端定义，将数据发送到executor执行累加，但是执行完累加后结果并没返回driver
    var counter2 = 0
    val rdd1: RDD[Int] = sc.parallelize(seq)
    rdd1.foreach(x => counter2 += x)
    println(counter2)
    //提出了在driver端和executor端共享当前变量
    //累加器也是在action操作的时候触发
    val acc: Accumulator[Int] = sc.accumulator(0)
    rdd1.foreach(x=>acc+=x)
    println(acc)
    //sc.accumulator该方法2.0已经弃用，可以直接使用sc.longAccumulator
    val acc_count: LongAccumulator = sc.longAccumulator("acc_count")
    rdd1.foreach(x=>acc_count.add(x))
    println(acc_count)//LongAccumulator(id: 51, name: Some(acc_count), value: 6)
    println(acc_count.value)
  }
}

package cn.spark.sparkbase.base

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

/**
 * DESC:
 * * 1-读取数据
 * * 2-使用广播变量，定义的一组非单词组合的list或map广播到executor执行
 * * 2-过滤出来，使用累加器统计非单词的组合
 * * 3-使用wordcout统计单词综合
 */
object _06acc_broadcast {
  def main(args: Array[String]): Unit = {
    //申请资源
    val sc: SparkContext = {
      val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")
      val sc = new SparkContext(conf)
      sc
    }
    // 1-读取数据
    //定义广播变量，因为每个单词都需要查看是否是费单词序列
    val list: List[String] = List(",", ".", "!", "#", "$", "%")
    val broadcastList: Broadcast[List[String]] = sc.broadcast(list)
    //定义累加器
    val acc_count: LongAccumulator = sc.longAccumulator("acc_count")

    val fileRDD: RDD[String] = sc.textFile("data/baseinput/words1.txt")
    val wordscount: RDD[String] = fileRDD
      .filter(line => line != null && line.trim.length > 0)
      .flatMap(line => line.split("\\s+"))
      .filter(word => {
        val listValue: List[String] = broadcastList.value
        val isFlag: Boolean = listValue.contains(word)
        if (isFlag) {
          acc_count.add(1L)
        }
        !isFlag
      })
    // 2-使用广播变量，定义的一组非单词组合的list或map广播到executor执行
    // 2-过滤出来，使用累加器统计非单词的组合
    // 3-使用wordcout统计单词综合
    println("wordcount的结果")
    wordscount.map((_, 1)).reduceByKey(_ + _).collect().foreach(println(_))
    println("非单词的组合：", acc_count.value)
  }
}

//刚开始的时候一直序列化不成功，应该是找不到类，java+scala混编，把scala设置成src了，去掉pom的这个配置就可以了。
//第一种方式
//实现一个KryoRegistrator注册类，在该类里面对自定义的序列化类进行注册，然后在conf里面配置该类
public class MyKryoRegistrator implements KryoRegistrator {
    @Override
    public void registerClasses( Kryo kryo) {
        kryo.register(User.class);
        kryo.register(User[].class);
        kryo.register(scala.collection.convert.Wrappers$.class);
    }
}
@Data
public class User implements Serializable {
        int id;
        String name;
        String city;
        List<String> hobby;
}

//  在conf配置如下
val conf = new SparkConf().setAppName(this.getClass.getSimpleName)
      conf
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .set("spark.kryo.registrationRequired", "true")
      .set("spark.kryo.registrator",classOf[MyKryoRegistrator].getName)
    

    val sc = new SparkContext(conf)

//第二种方式
 conf
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .registerKryoClasses(Array(classOf[User]))

下面列出Shuffle类不同分类算子
去重
def distinct()
def distinct(numPartitions: Int)
聚合
def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]
def groupBy[K](f: T => K, p: Partitioner):RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner):RDD[(K, Iterable[V])]
def aggregateByKey[U: ClassTag](zeroValue: U, partitioner: Partitioner): RDD[(K, U)]
def aggregateByKey[U: ClassTag](zeroValue: U, numPartitions: Int): RDD[(K, U)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, numPartitions: Int): RDD[(K, C)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null): RDD[(K, C)]
排序
def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]
def sortBy[K](f: (T) => K, ascending: Boolean = true, numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
重分区
def coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null)
集合或者表操作
def intersection(other: RDD[T]): RDD[T]
def intersection(other: RDD[T], partitioner: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
def intersection(other: RDD[T], numPartitions: Int): RDD[T]
def subtract(other: RDD[T], numPartitions: Int): RDD[T]
def subtract(other: RDD[T], p: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
def subtractByKey[W: ClassTag](other: RDD[(K, W)]): RDD[(K, V)]
def subtractByKey[W: ClassTag](other: RDD[(K, W)], numPartitions: Int): RDD[(K, V)]
def subtractByKey[W: ClassTag](other: RDD[(K, W)], p: Partitioner): RDD[(K, V)]
def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]
def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]