spark的简单学习一

一 RDD 

1.1 RDD的概述

1.RDD（Resilient Distributed Dataset，弹性分布式数据集）是Apache Spark中的一个核心概念。它是Spark中用于表示不可变、可分区、里面的元素可并行计算的集合。RDD提供了一种高度受限的共享内存模型，即RDD是只读的记录集的分区集合，只能通过在其他RDD执行确定的转换操作（如map、filter、join等）来创建新的RDD。

1.2 RDD的五大特性

  1. 不可变性（Immutable）：RDD 的内容不可修改，只能通过转换操作生成新的 RDD。

  2. 容错性（Fault-Tolerant）：RDD 可以容忍节点故障，通过存储每个分区的转换操作记录，可以在发生故障时重新计算丢失的分区。

  3. 可分区（Partitioned）：RDD 将数据划分为多个分区，这些分区可以在集群中的多个节点上并行处理。

  4. 可持久化（Persisted）：RDD 可以通过调用 persist() 或 cache() 方法将其缓存在内存中，以便在后续操作中重用。

  5. 惰性评估（Lazy Evaluation）：RDD 的转换操作是惰性的，只有在遇到行动操作时才会真正执行转换操作。

  • RDD由很多分区（partition）构成，有多少partition就对应有多少任务（task）

  • 算子实际上是作用在每一个分区上

  • RDD之间有依赖关系，宽依赖和窄依赖，用于切分Stage

  • Spark默认是hash分区，ByKey类的算子只能作用在kv格式的rdd上

  • Spark为task的计算提供了最佳的计算位置，移动计算而不是移动数据

1.3 依赖关系

1.宽依赖：父rdd对应多个子rdd

2.宽依赖的算子：

  • 所有byKey算子：如partitionBy、groupByKey、reduceByKey、aggregateByKey、foldByKey、combineByKey、sortByKey等。
  • repartition、cartesian算子。
  • 部分join算子，特别是非hash-partitioned的join算子。

3.窄依赖：一个父rdd或者多个父rdd对应一个子rdd

4.窄依赖算子：

map、filter、flatMap、mapPartitions、mapPartitionsWithIndex、sample、union、distinct、coalesce、repartitionAndUnion等。

5.切分stage，stage的个数等于宽依赖的个数+1

6.计算rdd之间的依赖关系构建DAG

7.窄依赖的分区数是不可以改变，取决于第一个RDD分区数，宽依赖可以在产生shuffle的算子上设置分区数

1.4 算子

1.4.1 map算子

1.map算子是处理里面每一个元素的

2.语法：

xxx.map((变量名：变量类型)=>{

        变量的处理

})

3.可以结合匹配选择，抽取元素

val clazzWithAgeRDD: RDD[(String, Int)] = splitRDD.map {case Array(_, _, age: String, _, clazz: String) => (clazz, age.toInt)}
   //创建spark环境val conf = new SparkConf()//明确运行模式conf.setMaster("local")//给任务取名字conf.setAppName("Map算子的演示")//创建对象val context: SparkContext = new SparkContext(conf)//读取文件val stuRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")//Map算子:将rdd中的数据，一条一条的取出来传入到map函数中，map会返回一个新的rdd，map不会改变总数据条数val mapRDD: RDD[List[String]] = stuRDD.map((stu: String) => {stu.split(",").toList})//使用foreach行动算子mapRDD.foreach(println)

1.4.2 filter

1.筛选的作用，返回另一个RDD

2.语法：

xxx.filter((变量名：变量类型)=>{

        变量的处理

})

/创建spark环境val conf = new SparkConf()//确定运行模式conf.setMaster("local")//给任务取名字conf.setAppName("filter算子演示")//创建对象val context: SparkContext = new SparkContext(conf)//读取文件val stuRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")/*** filter:过滤，将RDD中的数据一条一条取出传递给filter后面的函数，*        如果函数的结果是true，该条数据就保留，否则丢弃,取出来的结果组成一个新的RDD*///取出所有男生val filterRDD: RDD[String] = stuRDD.filter((stu: String) => {val stuList: Array[String] = stu.split(",")"男".equals(stuList(3))})filterRDD.foreach(println)//全都是男生的全部信息

1.4.3 flatMap

1.扁平化

2.语法：

xxx.flatMap((变量名：变量类型)=>{

        变量的处理

})

 //创建spark环境val conf = new SparkConf()//确定运行模式conf.setMaster("local")//给任务取名字conf.setAppName("flatMap算子演示")//创建对象val context: SparkContext = new SparkContext(conf)//读取文件val stuRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")/*** flatMap算子：将RDD中的数据一条一条的取出传递给后面的函数，*            函数的返回值必须是一个集合。最后会将集合展开构成一个新的RDD*///扁平化val value: RDD[String] = stuRDD.flatMap((stu: String) => {stu.split(",")})value.foreach(println)//结果是所有信息都是换行的

 1.4.4 sample

1.抽取RDD的部分

2.语法：

xxx.sample(withReplacement = true, 0-1的小数)

 //创建spark环境val conf = new SparkConf()//确定运行模式conf.setMaster("local")//给任务取名字conf.setAppName("sample算子演示")//创建对象val context: SparkContext = new SparkContext(conf)//读取文件val stuRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")/*** sample算子：从前一个RDD的数据中抽样一部分数据* 抽取的比例不是正好对应的，在抽取的比例上下浮动 比如1000条抽取10% 抽取的结果在100条左右*///withReplacement: Boolean, fraction: Double,val sampleRDD: RDD[String] = stuRDD.sample(withReplacement = true, 0.1)sampleRDD.foreach(println)

 1.4.5 groupBy

1.按照指定的字段进行分组，返回的是一个键是分组字段，值是一个存放原本数据的迭代器的键值对 返回的是kv格式的RDD

2.key: 是分组字段

 value: 是spark中的迭代器

 迭代器中的数据，不是完全被加载到内存中计算，迭代器只能迭代一次

 groupBy会产生shuffle

 //创建spark环境val conf = new SparkConf()//确定运行模式conf.setMaster("local")//给任务取名字conf.setAppName("groupBy算子演示")//创建对象val context: SparkContext = new SparkContext(conf)//读取文件val stuRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")val splitRDD: RDD[Array[String]] = stuRDD.map((stu: String) => {stu.split(",")})//求出每个班级的评价年龄//使用匹配模式//1、先取出班级和年龄val clazzWithAgeRDD: RDD[(String, Int)] = splitRDD.map {case Array(_, _, age: String, _, clazz: String) => (clazz, age.toInt)}//2、按照班级分组/*** groupBy：按照指定的字段进行分组，返回的是一个键是分组字段，* 值是一个存放原本数据的迭代器的键值对 返回的是kv格式的RDD** key: 是分组字段* value: 是spark中的迭代器* 迭代器中的数据，不是完全被加载到内存中计算，迭代器只能迭代一次** groupBy会产生shuffle*/val kvRDD: RDD[(String, Iterable[(String, Int)])] = clazzWithAgeRDD.groupBy((str: (String, Int)) => {str._1})//简写groupBy
//    val kvRDD: RDD[(String, Iterable[(String, Int)])] = clazzWithAgeRDD.groupBy((_._1))//(理科二班,CompactBuffer((理科二班,21), (理科二班,23), (理科二班,21), (理科二班,23)val clazzAvgAgeRDD: RDD[(String, Double)] = kvRDD.map((kv: (String, Iterable[(String, Int)])) => {val allAge: Iterable[Int] = kv._2.map((kv: (String, Int)) => {kv._2})(kv._1, allAge.sum.toDouble / allAge.size)})clazzAvgAgeRDD.foreach(println)

 1.4.6 groupByKey

1. groupByKey: 按照键进行分组，将value值构成迭代器返回

2.将来你在spark中看到RDD[(xx, xxx)] 这样的RDD就是kv键值对类型的RDD

3.只有kv类型键值对RDD才可以调用groupByKey算子

4.与groupBy的区别：

数据格式：groupBy不用考虑数据格式，而groupByKey必须是kv（键值对）数据格式。

分组规则：groupBy需要指定分组规则，即根据某个或某些字段进行分组；而groupByKey则是根据key对value进行分组。

返回值类型：groupBy是将整条数据放在集合中，即它会将数据集按照指定的规则划分成若干个小区域，并将这些小区域（包含整个数据行）作为集合返回；而groupByKey只是将具有相同key的value放在集合中，即它会把RDD的类型由RDD[(Key, Value)]转换为RDD[(Key, Value集合)]。

性能：groupByKey的性能更好，执行速度更快，因为groupByKey相比较与groupBy算子来说，shuffle所需要的数据量较少

//创建spark环境val conf = new SparkConf()//确定运行模式conf.setMaster("local")//给任务取名字conf.setAppName("groupByKey算子演示")//创建对象val context: SparkContext = new SparkContext(conf)//读取文件val stuRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")val splitRDD: RDD[Array[String]] = stuRDD.map((stu: String) => {stu.split(",")})//需求：求出每个班级平均年龄//使用模式匹配的方式取出班级和年龄val clazzWithAgeRDD: RDD[(String, Int)] = splitRDD.map {case Array(_, _, age: String, _, clazz: String) => (clazz, age.toInt)}/***  groupByKey: 按照键进行分组，将value值构成迭代器返回*  将来你在spark中看到RDD[(xx, xxx)] 这样的RDD就是kv键值对类型的RDD*  只有kv类型键值对RDD才可以调用groupByKey算子**/val kvRDD: RDD[(String, Iterable[Int])] = clazzWithAgeRDD.groupByKey()//(理科六班,CompactBuffer(22, 22, 23, 22, 21, 24, 21, 21, 22))val clazzAvgAgeRDD: RDD[(String, Double)] = kvRDD.map((kv: (String, Iterable[(Int)])) => {(kv._1,kv._2.sum.toDouble/kv._2.size)})clazzAvgAgeRDD.foreach(println)

1.4.7 reduceByKey

1.利用reduceByKey实现：按照键key对value值直接进行聚合，需要传入聚合的方式 

2.reduceByKey算子也是只有kv类型的RDD才能调用

3.与groupByKey的区别

  1. 功能：

    • reduceByKey：该函数用于对具有相同键的值进行聚合操作。它会将具有相同键的值按照指定的合并函数进行迭代和聚合，最终生成一个新的RDD，其中每个键都是唯一的，与每个键相关联的值是经过合并操作后的结果。
    • groupByKey：该函数仅根据键对RDD中的元素进行分组，不执行任何聚合操作。它只是将具有相同键的元素放在一个组中，形成一个包含键和其对应值的迭代器。因此，groupByKey的结果是一个新的RDD，其中每个键都与一个迭代器相关联，迭代器包含了与该键关联的所有值。

  2. 结果：

    • reduceByKey：返回一个新的RDD，其中每个键都是唯一的，与每个键相关联的值是经过合并操作后的结果。
    • groupByKey：返回一个新的RDD，其中每个键都与一个迭代器相关联，迭代器包含了与该键关联的所有值。

  3. 性能：

    • reduceByKey：在某些情况下可能更高效，因为它可以在分布式计算中在map阶段进行一些本地聚合，从而减少数据传输。
    • groupByKey：可能导致数据移动较多，因为它只是对键进行分组，而不进行本地聚合。因此，在处理大数据集时，groupByKey可能会导致更高的网络传输成本和更长的处理时间。

4.以后遇见key相同，value相加直接用reduceByKey

xxx.reduceByKey((x:Int,y:Int)=>x+y)

   //创建spark环境val conf = new SparkConf()//确定运行模式conf.setMaster("local")//给任务取名字conf.setAppName("reduceByKey算子演示")//创建对象val context: SparkContext = new SparkContext(conf)//读取文件val stuRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")val splitRDD: RDD[Array[String]] = stuRDD.map((stu: String) => {stu.split(",")})//求每个班级的人数//1、将每个元素变成(clazz,1)val mapRDD: RDD[(String, Int)] = splitRDD.map {case Array(_, _, _, _, clazz: String) => (clazz, 1)}/*** 利用reduceByKey实现：按照键key对value值直接进行聚合，需要传入聚合的方式* reduceByKey算子也是只有kv类型的RDD才能调用*///聚合val clazzSumPersonRDD: RDD[(String, Int)] = mapRDD.reduceByKey((x: Int, y: Int) => x + y)clazzSumPersonRDD.foreach(println)

1.4.8 union

1.上下合并两个RDD,前提是两个RDD中的数据类型要一致，合并后不会对结果进行去重

注：这里的合并只是逻辑层面上的合并，物理层面其实是没有合并

val conf = new SparkConf()conf.setMaster("local")conf.setAppName("Union算子演示")val context = new SparkContext(conf)//====================================================val w1RDD: RDD[String] = context.textFile("spark/data/ws/w1.txt") // 1val w2RDD: RDD[String] = context.textFile("spark/data/ws/w2.txt") // 1/***  union:上下合并两个RDD,前提是两个RDD中的数据类型要一致，合并后不会对结果进行去重**  注：这里的合并只是逻辑层面上的合并，物理层面其实是没有合并*/val unionRDD: RDD[String] = w1RDD.union(w2RDD)unionRDD.foreach(println)

 1.4.9 join

1.内连接，左连接，右连接，全连接

2.连接的2个RDD里面数据类型必须一样

 val conf = new SparkConf()conf.setMaster("local")conf.setAppName("Join算子演示")val context = new SparkContext(conf)//====================================================//两个kv类型的RDD之间的关联//通过scala中的集合构建RDD,通过context中的parallelize方法将集合变成RDDval rdd1: RDD[(String, String)] = context.parallelize(List("1001" -> "hjx","1002" -> "hdx","1003" -> "hfx","1004" -> "hhx","1005" -> "hkx","1007" -> "hbx"))val rdd2: RDD[(String, String)] = context.parallelize(List(("1001", "崩坏"),("1002", "原神"),("1003", "王者"),("1004", "修仙"),("1005", "学习"),("1006", "敲代码")))/*** 内连接：join* 左连接：leftJoin* 右连接：rightJoin* 全连接：fullJoin*///join
//    val joinRDD: RDD[(String, (String, String))] = rdd1.join(rdd2)
//    joinRDD.foreach(println)//(1005,(hkx,学习))//leftJoin
//    val leftRDD: RDD[(String, (String, Option[String]))] = rdd1.leftOuterJoin(rdd2)
//    val leftRDD1: RDD[(String, String, String)] = leftRDD.map {
//      case (id: String, (name: String, Some(like))) => (id, name, like)
//      case (id: String, (name: String, None)) => (id, name, "没有爱好")
//    }
//    leftRDD1.foreach(println)//rightJoin
//    val rightRDD: RDD[(String, (Option[String], String))] = rdd1.rightOuterJoin(rdd2)
//    val rightRDD1: RDD[(String, String, String)] = rightRDD.map {
//      case (id: String, (Some(name), like: String)) => (id, name, like)
//      case (id: String, (None, like: String)) => (id, "查无此人", like)
//    }
//    rightRDD1.foreach(println)//fullJoinval fullJoin: RDD[(String, (Option[String], Option[String]))] = rdd1.fullOuterJoin(rdd2)val fullJoin1: RDD[(String, String, String)] = fullJoin.map {case (id: String, (Some(name), Some(like))) => (id, name, like)case (id: String, (None, Some(like))) => (id, "查无此人", like)case (id: String, (Some(name), None)) => (id, name, "没有爱好")}fullJoin1.foreach(println)

1.4.10 sortBy

1.返回的也是一个RDD

//创建spark环境val conf = new SparkConf()//明确运行模式conf.setMaster("local")//给任务取名字conf.setAppName("统计总分年级排名前10的学生的各科分数")//创建对象val context: SparkContext = new SparkContext(conf)//需求：统计总分年级排名前10的学生的各科分数//1、读取文件val mapRDD: RDD[(String, String, Int)] = context.textFile("spark/data/ws/score.txt").map((s: String) => {s.split(",")}) //切分数据.filter((array: Array[String]) => {array.length == 3})//过滤数据.map {case Array(sid: String, subject_id, score: String) => (sid, subject_id, score.toInt)}//筛选数据//2、计算每个学生的总分val mapRDD1: RDD[(String, Int)] = mapRDD.map {//RDD中的模式匹配，case后面不需要加类型，直接是RDD小括号中的数据类型匹配case (sid: String, _, score: Int) => (sid, score)}//以后遇见key相同，value相加，直接用reduceByKey。reduceByKey((x: Int, y: Int) => x + yval reduceRDD: RDD[(String, Int)] = mapRDD1.reduceByKey((x: Int, y: Int) => x + y)//3、按照总分排序val sortByRDD: Array[(String, Int)] = reduceRDD.sortBy((kv: (String, Int)) => {-kv._2}).take(10)//4、求各科成绩//拿出学号val ids: Array[String] = sortByRDD.map((kv: (String, Int)) => {kv._1})val clazzScoreTop10RDD: RDD[(String, String, Int)] = mapRDD.filter {case (sid: String, _, _) => ids.contains(sid)}clazzScoreTop10RDD.foreach(println)

1.4.11 mapValues

1. mapValues算子：也是作用在kv类型的RDD上

2.主要的作用键不变，处理值

    //需求：统计总分年级排名前10的学生的各科分数//1、读取文件val mapRDD: RDD[(String, String, Int)] = context.textFile("spark/data/ws/score.txt").map((s: String) => {s.split(",")}) //切分数据.filter((array: Array[String]) => {array.length == 3})//过滤数据.map {case Array(sid: String, subject_id, score: String) => (sid, subject_id, score.toInt)}//筛选数据//2、计算每个学生的总分val mapRDD1: RDD[(String, Int)] = mapRDD.map {//RDD中的模式匹配，case后面不需要加类型，直接是RDD小括号中的数据类型匹配case (sid: String, _, score: Int) => (sid, score)}//以后遇见key相同，value相加，直接用reduceByKey。reduceByKey((x: Int, y: Int) => x + yval reduceRDD: RDD[(String, Int)] = mapRDD1.reduceByKey((x: Int, y: Int) => x + y)/*** mapValues算子：也是作用在kv类型的RDD上* 主要的作用键不变，处理值*/val mapValuesRDD: RDD[(String, Int)] = reduceRDD.mapValues(_ + 1000)//等同于val mapValues1: RDD[(String, Int)] = reduceRDD.map((kv: (String, Int)) => {(kv._1, kv._2 + 1000)})

1.4.12 mapPartitions

1.mapPartitions与mapPartitionsWithIndex的用法

mapPartitions：不用指定分区，里面传入的是迭代器，迭代器存储的是每个分区的数据

mapPartitionsWithIndex：指定分区，

 //创建spark环境val conf = new SparkConf()//明确运行模式conf.setMaster("local")//给任务取名字conf.setAppName("mapPartition算子的演示")//创建对象val context: SparkContext = new SparkContext(conf)val scoreRDD: RDD[String] = context.textFile("spark/data/ws/*") // 读取数据文件//打印分区println(scoreRDD.getNumPartitions)//
//    val mapPartition: RDD[String] = scoreRDD.mapPartitions((itr: Iterator[String]) => {
//      itr.flatMap((s: String) => {
//        s.split("\\|")
//      })
//    })
//    mapPartition.foreach(println)
//    scoreRDD.mapPartitionsWithIndex{
//      case (index:Int,itr:Iterator[String])=>
//        println(s"当成分区${index}")
//        itr.flatMap((s:String)=>{
//          s.split("\\|")
//        })
//    }.foreach(println)val mapPartitionRDD: RDD[String] = scoreRDD.mapPartitionsWithIndex((i: Int, itr: Iterator[String]) => {println(s"分区是${i}")itr.flatMap(_.split("\\|"))})mapPartitionRDD.foreach(println)

1.4.13 行动算子

1.一个行动算子，执行一次任务。没有行动算子，RDD没有结果

2.执行顺序：除去算子的操作先执行，再执行RDD里面的操作

3.collcet：将RDD变成scala中的集合

  //创建spark环境val conf = new SparkConf()//明确运行模式conf.setMaster("local")//给任务取名字conf.setAppName("action算子")//创建对象val context: SparkContext = new SparkContext(conf)//读取文件val stuRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")println("hhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhhh")//val studentsRDD: RDD[(String, String, String, String, String)] = stuRDD.map(_.split(",")).map {case Array(id: String, name: String, age: String, gender: String, clazz: String) =>println("jjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjjj")(id, name, age, gender, clazz)}println("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")studentsRDD.foreach(println)

二 spark流程

2.1 任务流程

1.读文件

2.切分/筛选

3.筛选

4.分组聚合

5.筛选

6.写入文件

2.2 缓存

2.2.1 catch算子

1.默认将数据缓存在内存中，程序结束，缓存数据没了

xxxRDD.catch()

2.2.2 persist算子

1.默认将数据缓存在内存中，catch的实际就是用的persist，程序结束，缓存数据没了

2.可以设置级别

xxxRDD.persist(StorageLevel.级别)

2.2.3 checkPoint                           

1.可以将RDD运行时的数据永久持久化在HDFS上，这个方案叫做checkpoint,需要在spark环境中设置checkpoint的路径

2.这个不能写在一个程序的末尾，要不然还是没有数据

 //创建spark环境val conf = new SparkConf()//明确运行模式conf.setMaster("local")//给任务取名字conf.setAppName("persist")//创建对象val context: SparkContext = new SparkContext(conf)//设置缓冲路径context.setCheckpointDir("spark/data/checkpoint")//读取文件val stuRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")//切分，筛选元素val studentsRDD: RDD[(String, String, String, String, String)] = stuRDD.map(_.split(",")).map {case Array(id: String, name: String, age: String, gender: String, clazz: String) =>(id, name, age, gender, clazz)}//统计每个班的人数studentsRDD.checkpoint()val clazzSumRDD: RDD[(String, Int)] = studentsRDD.map {case (_, _, _, _, clazz: String) => (clazz, 1)}.reduceByKey((x: Int, y: Int) => x + y) //可以简写为(_+_)clazzSumRDD.saveAsTextFile("spark/data/clazz_num")//统计男生女有多少人val genderSumRDD: RDD[(String, Int)] = studentsRDD.map {case (_, _, _, gender: String, _) => (gender, 1)}.reduceByKey((x: Int, y: Int) => x + y) //可以简写为(_+_)clazzSumRDD.saveAsTextFile("spark/data/gender_num")//def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)//    clazzSumRDD.persist(StorageLevel.MEMORY_ONLY)

三 spark部署环境

3.1 Standalone

3.1.1 解压

tar -zxvf 文件名 -C 路径

3.1.2 配置文件

1.环境变量

/etc/profile

2.复制一份模板，配置spark-env.sh文件

export SPARK_MASTER_IP=master
export SPARK_MASTER_PORT=7077export SPARK_WORKER_CORES=2
export SPARK_WORKER_INSTANCES=1
export SPARK_WORKER_MEMORY=2g
export JAVA_HOME=/usr/local/soft/jdk1.8.0_171

3.复制一份模板配置workers文件

node1
node2

3.1.3 复制

1复制一份给其他节点

3.1.4启动

1.本地集群模式启动进入sbin目录下启动

./start-all.sh

看是否启动成功，输入网址master:8080，有页面即是成功。

3.1.5 提交spark任务

1.client模式提交：

进入spark/example/jars目录下输入

spark-submit --class org.apache.spark.examples.SparkPi --master spark://master:7077 --executor-memory 512m --total-executor-cores 1 spark-examples_2.12-3.1.3.jar 100

spark-examples_2.12-3.1.3.jar：是jars下的jar名

spark-submit：命令的名字

class org.apache.spark.examples.SparkPi：类名

2.cluster模式

进入spark/example/jars目录下输入

spark-submit --class org.apache.spark.examples.SparkPi --master spark://master:7077 --executor-memory 512M --total-executor-cores 1 --deploy-mode cluster spark-examples_2.12-3.1.3.jar 100

3.1.6 实例

//创建spark环境val conf = new SparkConf()//提交Linux运行不需要明确运行模式
//    conf.setMaster("local")//给任务取个名字conf.setAppName("Standalone运行模式")//创建对象val sparkContext = new SparkContext(conf)//使用对象中的parallelize方法将Scala中的集合变成RDDval arrayRDD: RDD[String] = sparkContext.parallelize((List("java,hello,world","hello,scala,spark","java,hello,spark")))val flatMapRDD: RDD[String] = arrayRDD.flatMap(_.split(","))val mapRDD: RDD[(String, Int)] = flatMapRDD.map((s: String) => (s, 1))val reduceRDD: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)reduceRDD.foreach(println)/***  将项目打包放到spark集群中使用standalone模式运行* standalone client* spark-submit --class com.shujia.core.Demo17SparkStandaloneSubmit --master spark://master:7077 --executor-memory 512m --total-executor-cores 1 spark-1.0.jar 100** standalone cluster* spark-submit --class com.shujia.core.Demo17SparkStandaloneSubmit --master spark://master:7077 --executor-memory 512m --total-executor-cores 1 --deploy-mode cluster spark-1.0.jar 100**/

 1.将这个打包，放到Linux中

2.client模式提交：

spark-submit --class com.shujia.core.Demo17SparkStandaloneSubmit --master spark://master:7077 --executor-memory 512m --total-executor-cores 1 spark-1.0.jar 100

   这里就能看见结果

3.cluster模式：

先把jar包复制给子节点，再提交任务

spark-submit --class com.shujia.core.Demo17SparkStandaloneSubmit --master spark://master:7077 --executor-memory 512m --total-executor-cores 1 --deploy-mode cluster spark-1.0.jar 100

结果跟上面的看法一样

3.2 YARN

3.2.1配置文件

1修改spark-env.sh文件

export HADOOP_CONF_DIR=/usr/local/soft/hadoop-3.1.1/etc/hadoop

2.修改yarn-site.xml

<property>
<name>yarn.nodemanager.pmem-check-enabled</name>
<value>false</value>
</property><property>
<name>yarn.nodemanager.vmem-check-enabled</name>
<value>false</value>
</property><property>
<name>yarn.application.classpath</name>
<value>(在master输入hadoop classpath，将那段话复制过来，不要有空格)</value>
</property>

3.同步到其他节点

3.2.2 提交任务

1.yarn-client提交

spark-submit --master yarn --deploy-mode client --class org.apache.spark.examples.SparkPi  spark-examples_2.12-3.1.3.jar 100

spark-submit：命令名字

deploy-mode client：啥类型提交

class org.apache.spark.examples.SparkPi：类名

spark-examples_2.12-3.1.3.jar 100：jar包名

运行大概流程

2.yarn-cluster提交

spark-submit --master yarn --deploy-mode cluster --class org.apache.spark.examples.SparkPi  spark-examples_2.12-3.1.3.jar 100

查看结果：yarn logs -applicationId xxxxxxxxxx

四 其他知识

4.1 资源调度与任务调度

所说的基于spark-yarn client提交任务

4.1.1 资源调度     

1.本地启动Driver

2.向yarn(ResourceManager)申请资源 ，提交spark Application

3.RM接收spark Application过后，会分配一个子节点启动ApplicationMaster进程 

4.ApplicationMaster向RM申请节点并启动Executor 

5.Executor反向注册给Driver

4.1.2 任务调度

1.当代码遇见一个action算子，开始进行任务调度

2.Driver根据RDD之间的依赖关系将Application形成一个DAG（有向无环图）。

4.将DAG发送给DAG Scheduler

3.DAG Scheduler会根据产生的shuffle划分窄宽依赖，通过宽依赖划分Stage

4.DAG Scheduler将Stage包装成taskset发送给Task Scheduler(stage里面有很多并行的task，taskset是每个stage里面的并行task封装的)

5.Task Scheduler拿到了task后发送到Executor中的线程池执行

4.1.3 重试机制

1.如果task执行失败taskscheduler会重试3次，如果还失败，DAGscheduler会重试4次

2.存在不会重试的场景

如果是因为shuffle过程中文件找不到的异常，taskscheduler不负责重试task,而是由DAGscheduler重试上一个stage

4.1.4推行执行

1.如果有的task执行很慢，taskscheduler会在发生一个一摸一样的task到其它节点中执行，让多个task竟争，谁先执行完成以谁的结果为准

4.2 累加器

1.RDD内部的改变不会影响RDD外面的计算

 val conf = new SparkConf()conf.setMaster("local")conf.setAppName("累加器演示")val context = new SparkContext(conf)//====================================================val studentRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")var count = 0studentRDD.foreach((line: String) => {count += 1println("-------------------------")println(count)println("-------------------------")})println(s"count的值为：${count}")//0

上述这个程序RDD里面的count输出是1000，而RDD外面的count还是0

2.实现累加器(触发作业执行之后加的) 

  def main(args: Array[String]): Unit = {val conf = new SparkConf()conf.setMaster("local")conf.setAppName("累加器演示")val context = new SparkContext(conf)//====================================================val studentRDD: RDD[String] = context.textFile("spark/data/ws/students.csv")//    var count = 0
//    studentRDD.foreach((line: String) => {
//      count += 1
//      println("-------------------------")
//      println(count)
//      println("-------------------------")
//    })
//    println(s"count的值为：${count}")//0/*** 累加器** 由SparkContext来创建* 注意：*  1、因为累加器的执行实在RDD中执行的，而RDD是在Executor中执行的，而要想在Executor中执行就得有一个action算子触发任务调度*  2、sparkRDD中无法使用其他的RDD*  3、SparkContext无法在RDD内部使用，因为SparkContext对象无法进行序列化，不能够通过网络发送到Executor中*/val longAccumulator: LongAccumulator = context.longAccumulatorstudentRDD.foreach((line:String)=>{longAccumulator.add(1)})println(longAccumulator.value)

4.3 广播变量

1.避免了每次Task任务拉取都要附带一个副本，拉取的速度变快了，执行速度也就变快了

2.未使用广播变量

//创建spark环境val conf = new SparkConf()//明确模式conf.setMaster("local")//给任务取个名字conf.setAppName("广播变量")//创建对象val context = new SparkContext(conf)//以scala的方式读取students.csv文件并进行相关操作val stuMap: Map[String, String] = Source.fromFile("spark/data/ws/students.csv").getLines().toList.map((line: String) => {val lines: Array[String] = line.split(",")(lines(0), lines.mkString(","))}).toMap//使用spark的形式读取文件val scoreRDD: RDD[String] = context.textFile("spark/data/ws/score.txt")/*** 将Spark读取的分数RDD与外部变量学生Map集合进行关联* 循环遍历scoresRDD，将学号一样的学生信息关联起来*/val idWithInfosRDD: RDD[(String, String)] = scoreRDD.map((score: String) => {val id: String = score.split(",")(0)val infos: String = stuMap.getOrElse(id, "查无此人")(id, infos)})idWithInfosRDD.foreach(println)

3.使用广播变量

def main(args: Array[String]): Unit = {//创建spark环境val conf = new SparkConf()//明确模式conf.setMaster("local")//给任务取个名字conf.setAppName("广播变量")//创建对象val context = new SparkContext(conf)//以scala的方式读取students.csv文件并进行相关操作val stuMap: Map[String, String] = Source.fromFile("spark/data/ws/students.csv").getLines().toList.map((line: String) => {val lines: Array[String] = line.split(",")(lines(0), lines.mkString(","))}).toMap//使用spark的形式读取文件val scoreRDD: RDD[String] = context.textFile("spark/data/ws/score.txt")/*** 将Spark读取的分数RDD与外部变量学生Map集合进行关联* 循环遍历scoresRDD，将学号一样的学生信息关联起来*/
//    val idWithInfosRDD: RDD[(String, String)] = scoreRDD.map((score: String) => {
//      val id: String = score.split(",")(0)
//      val infos: String = stuMap.getOrElse(id, "查无此人")
//      (id, infos)
//    })//创建广播变量val stuBro: Broadcast[Map[String, String]] = context.broadcast(stuMap)val value: RDD[(String, String)] = scoreRDD.map((score: String) => {val id: String = score.split(",")(0)val map: Map[String, String] = stuBro.valueval infos: String = map.getOrElse(id, "ddd")(id, infos)})value.foreach(println)

4.4 blockmanager

                                                                

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/839843.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

IDEA连接MySQL后如何管理数据库

IDEA连接MySQL后如何管理数据库

上一节讲解了IDEA如何连接MySQL数据库管理系统&#xff0c;接下来我们就可以在IDEA里使用MySQL来管理数据库了。那么如果我们现在还没有创建需要的数据库怎么办&#xff1f;本节就来教大家如何在IDEA连接MySQL后管理数据库(创建/修改/删除数据库、创建/修改/删除表、插入/更新/…
阅读更多...
电子招投标系统源码实现与立项流程：基于Spring Boot、Mybatis、Redis和Layui的企业电子招采平台

电子招投标系统源码实现与立项流程：基于Spring Boot、Mybatis、Redis和Layui的企业电子招采平台

随着企业的快速发展&#xff0c;招采管理逐渐成为企业运营中的重要环节。为了满足公司对内部招采管理提升的要求&#xff0c;建立一个公平、公开、公正的采购环境至关重要。在这个背景下&#xff0c;我们开发了一款电子招标采购软件&#xff0c;以最大限度地控制采购成本&#…
阅读更多...
【Vue2.x】props技术详解

【Vue2.x】props技术详解

1.什么是prop&#xff1f; 定义&#xff1a;组件标签上注册的一些自定义属性作用&#xff1a;向子组件传递数据特点 可以传递任意数量的prop可以传递任意类型的prop 2.prop校验 为了避免乱传数据&#xff0c;需要进行校验 完整写法 将之前props数组的写法&#xff0c;改为对象…
阅读更多...
【SQL Server001】SQLServer2016常用函数实战总结（已更新）

【SQL Server001】SQLServer2016常用函数实战总结（已更新）

1.熟悉、梳理、总结下SQL Server相关知识体系。 2.日常研发过程中使用较少&#xff0c;随着时间的推移&#xff0c;很快就忘得一干二净&#xff0c;所以梳理总结下&#xff0c;以备日常使用参考 3.欢迎批评指正&#xff0c;跪谢一键三连&#xff01; 总结源文件资源下载地址&am…
阅读更多...
Ubuntu切换内核版本

Ubuntu切换内核版本

#安装内核安装工具 sudo apt-get install software-properties-common sudo add-apt-repository ppa:cappelikan/ppa sudo apt-get update sudo apt-get install mainline#安装指定内核版本(有些版本并不能安装成功) mainline install 5.14.10#更新GRUB配置 sudo update-grub#查…
阅读更多...
PE文件（六）新增节-添加代码

PE文件（六）新增节-添加代码

本节的目的是在所有节的空白区都不够存放我们要添加的代码时&#xff0c;教会我们新增一个足够大的节来添加代码 添加节 一.判断是否有足够的空间可以添加一个节表&#xff1a;新增节需要新增一个节表来记录此节信息 判断方法&#xff1a;SizeOfHeader - (DOS 垃圾数据 PE…
阅读更多...
全网最全爬取-b站爬取弹幕+评论之js逆向与xml降本增效

全网最全爬取-b站爬取弹幕+评论之js逆向与xml降本增效

&#x1f31f; ❤️ 作者&#xff1a;yueji0j1anke 首发于公号&#xff1a;剑客古月的安全屋 字数&#xff1a;801 阅读时间: 10min 声明&#xff1a;请勿利用文章内的相关技术从事非法测试&#xff0c;由于传播、利用此文所提供的信息而造成的任何直接或者间接的后果及…
阅读更多...
【C语言深度解剖】(14)：结构体内存对齐（详细配图讲解）

【C语言深度解剖】(14)：结构体内存对齐（详细配图讲解）

&#x1f921;博客主页&#xff1a;醉竺 &#x1f970;本文专栏&#xff1a;《C语言深度解剖》 &#x1f63b;欢迎关注&#xff1a;感谢大家的点赞评论关注&#xff0c;祝您学有所成&#xff01; ✨✨&#x1f49c;&#x1f49b;想要学习更多C语言深度解剖点击专栏链接查看&…
阅读更多...
缩进在编程中的重要性及正确使用方法

缩进在编程中的重要性及正确使用方法

新书上架~&#x1f447;全国包邮奥~ python实用小工具开发教程http://pythontoolsteach.com/3 欢迎关注我&#x1f446;&#xff0c;收藏下次不迷路┗|&#xff40;O′|┛ 嗷~~ 目录 前言 缩进不当引发的问题 缩进的正确使用方法 缩进错误的调试与修复 总结 前言 在编程世…
阅读更多...
Unity 资源 之 限时免费的Lowpoly农场动物，等你来领！

Unity 资源 之 限时免费的Lowpoly农场动物，等你来领！

Unity资源 之 Lowpoly farm animals 农村动物 前言资源包内容领取兑换码 前言 Unity 资源商店为大家带来了一份特别的惊喜——限时免费的农场动物资源&#xff01;这是一个充满趣味和实用性的资源包。 资源包内容 在这个资源包中&#xff0c;你可以找到丰富多样的低地养殖动物…
阅读更多...
Vue3路由配置

Vue3路由配置

路由其实就是一组对应关系&#xff0c;将一个路径与一个组件对应起来&#xff0c;当路径发生变化&#xff0c;路由器就可以通过路由规则&#xff0c;找到当前路径对应的组件&#xff0c;并将该组件呈现到页面上 使用路由步骤&#xff1a; 1.终端输入 npm i vue-router 2.在App…
阅读更多...
Softing工业将亮相2024年阿赫玛展会——提供过程自动化的连接解决方案

Softing工业将亮相2024年阿赫玛展会——提供过程自动化的连接解决方案

您可于2024年6月10日至14日前往美因河畔法兰克福11.0号馆&#xff0c;Softing将在C25展位展出&#xff0c;欢迎莅临&#xff01; 作为工业应用中数据交换领域公认的专家&#xff0c;Softing工业致力于帮助各行各业的客户部署网络自动化和优化生产流程。 使用Softing产品&…
阅读更多...
如何在OpenHarmony上使用SeetaFace2人脸识别库？

如何在OpenHarmony上使用SeetaFace2人脸识别库？

简介 相信大部分同学们都已了解或接触过OpenAtom OpenHarmony&#xff08;以下简称“OpenHarmony”&#xff09;了&#xff0c;但你一定没在OpenHarmony上实现过人脸识别功能&#xff0c;跟着本文带你快速在OpenHarmony标准设备上基于SeetaFace2和OpenCV实现人脸识别。 项目效…
阅读更多...
【Vue】Vue2路由

【Vue】Vue2路由

目录 路由作用Vue Router路由Vue Router路由的组成VueRouter常用的函数Vue Router的使用安装Vue Router创建router引入router使用 备注 Vue多级路由&#xff08;嵌套路由&#xff09;编写组件配置嵌套路由 Vue中的动态路由代码示例父组件Home.vue子组件路由配置 路由的 query 参…
阅读更多...
黑龙江等保测评深入理解

黑龙江等保测评深入理解

“没有网络安全&#xff0c;就没有国家安全”&#xff0c;等级保护测评是指按照网络安全系统制定的一系列的防护过程&#xff0c;对已经有的和即将上线的商业服务的基础设施&#xff08;系统&#xff0c;数据库&#xff0c;中间件等&#xff09;所做的一系列的检查&#xff0c;…
阅读更多...
HeyGen AI是什么？怎样使用HeyGen AI？

HeyGen AI是什么？怎样使用HeyGen AI？

在数字时代&#xff0c;视频内容为王。无论是在社交媒体还是网站上&#xff0c;视频都以其独特的方式吸引着人们的眼球。然而&#xff0c;制作出专业水准的视频往往需要大量的时间和技术知识。HeyGen AI正是为了解决这一难题而诞生的。 HeyGen AI简介 HeyGen AI是一个创新的视…
阅读更多...
618值得买的好物清单，这些数码好物你千万不能错过！

618值得买的好物清单，这些数码好物你千万不能错过！

​随着618购物节的距离越来越近&#xff0c;你是不是已经开始疯狂浏览购物app&#xff0c;准备大肆采购一番了&#xff1f;但是在购物之前&#xff0c;还是得先做一做功课&#xff0c;避免陷入购物陷阱&#xff0c;而作为一名经验丰富的数码爱好者&#xff0c;想通过这次机会给…
阅读更多...
Thinkphp内核开发盲盒商城源码v2.0 对接易支付/阿里云短信/七牛云存储

Thinkphp内核开发盲盒商城源码v2.0 对接易支付/阿里云短信/七牛云存储

源码简介 这套系统是我从以前客户手里拿到的,100完整可用,今天测试防红链接失效了,需要修改防红API即可!前端页面展示我就不放了,懂的都懂 优点是Thinkphp开发的&#xff0c;二开容易。 源码图片 资源获取&#xff1a;Thinkphp内核开发盲盒商城源码v2.0 对接易支付/阿里云短…
阅读更多...
kafka监控配置和告警配置——筑梦之路

kafka监控配置和告警配置——筑梦之路

kafka_exporter项目地址&#xff1a;https://github.com/danielqsj/kafka_exporter docker-compose部署kafka_exporter # docker-compose部署多个kafka_exporter&#xff0c;每个exporter对接一个kafka# cat docker-compose.ymlversion: 3.1 services:kafka-exporter-opslogs…
阅读更多...
平安银行广州分行护航客户子女升学之路，规划美好未来

平安银行广州分行护航客户子女升学之路，规划美好未来

阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023