spark的简单学习一

news/2024/7/14 19:50:20/文章来源:https://blog.csdn.net/weixin_65909965/article/details/139121703

一 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.ApplicationMasterRM申请节点并启动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 SchedulerStage包装成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.luyixian.cn/news_show_1052891.aspx

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

相关文章

如何基于springboot构建cas最新版源码?

环境准备 下载JDK21 https://download.oracle.com/java/21/archive/jdk-21.0.2_windows-x64_bin.zip下载gradle 8.5并配置环境变量 https://gradle.org/next-steps/?version8.5&formatbin下载项目git clone http://gitlab.ruishan.cc/meta/anka-authentication.git 开始…

【CSP CCF记录】201909-1 小明种苹果

题目 过程 #include<bits/stdc.h> using namespace std; int N,M; long long tree[1010]; int main() {cin>>N>>M;long long result0,max0;//result剩余苹果&#xff0c;max最大疏果个数 int id0;//id最大疏果的果树编号 for(int i1;i<N;i){long long b0…

【LeetCode】【9】回文数(1047字)

文章目录 [toc]题目描述样例输入输出与解释样例1样例2样例3 提示进阶Python实现 个人主页&#xff1a;丷从心 系列专栏&#xff1a;LeetCode 刷题指南&#xff1a;LeetCode刷题指南 题目描述 给一个整数x&#xff0c;如果x是一个回文整数&#xff0c;返回true&#xff1b;否…

Java数组的使用

Java数组的使用 前言一、数组基本用法什么是数组注意事项创建数组基本语法代码示例注意事项 数组的使用代码示例获取长度 & 访问元素注意事项 下标越界遍历数组编程求平均成绩Mathrandom类现有100个学生&#xff0c;编程求平均成绩 使用 for-each 遍历数组 二、数组作为方法…

解线性方程组——最速下降法及图形化表示 | 北太天元 or matlab

一、思路转变 A为对称正定矩阵&#xff0c; A x b Ax b Axb 求解向量 x x x这个问题可以转化为一个求 f ( x ) f(x) f(x)极小值点的问题&#xff0c;为什么可以这样&#xff1a; f ( x ) 1 2 x T A x − x T b c f(x) \frac{1}{2}x^TAx - x^Tb c f(x)21​xTAx−xTbc 可…

tty/pty/console/getty/shell/telnet

tty 终端(termimal)= tty(Teletypewriter, 电传打印机),作用是提供一个命令的输入输出环境,在linux下使用组合键ctrl+alt+T打开的就是终端,可以认为terminal和tty是同义词。 tty泛指所有的终端设置,这些是真实存在的设备。 通过tty命令可以查看当前终端连接的设备。…

Linux 一键部署alfresco 6

alfresco 前言 Alfresco是一个流行的企业级开源内容管理系统和协作平台。它提供了丰富的功能,包括文档管理、记录管理、协作工具、工作流管理、搜索和版本控制等。Alfresco还具有灵活的部署选项,可以作为本地部署的软件或云服务来使用。 该平台可以帮助组织管理和存储各种类…

WPS文件没有保存怎么恢复?5个解决方案轻松恢复!

“我在WPS上编辑了一个文件&#xff0c;但是还没来得及将它保存&#xff0c;我不小心就退出软件了&#xff0c;现在不知道有什么方法可以恢复WPS文件呢&#xff1f;大家可以帮帮我吗” WPS作为一款功能强大且用户友好的软件&#xff0c;给我们的工作带来了很多的便利。但我们在…

适用于Android的最佳数据恢复软件

如果您的 Android 设备崩溃&#xff0c;您需要找到一种方法来取回您的数据。幸运的是&#xff0c;有许多数据恢复程序可以帮助您恢复丢失的文件。有些是免费的&#xff0c;而另一些则需要付费。这是适用于Android设备的最佳数据恢复软件列表。 什么是数据恢复软件&#xff1f; …

紫光展锐前沿探索 | 满足未来6G多差异化应用场景的技术体系思考

在6G架构/系统设计中&#xff0c;紫光展锐提出了未来6G空口“一体多翼”的技术体系概念&#xff0c;即“Big-Lite Multi-RAT”。本文将详细对该技术体系展开介绍。 “一体多翼”技术体系通过 “体”&#xff08;Big RAT&#xff09;和“翼”&#xff08;Lite RAT&#xff09;的…

Java语言ADR药物不良反应系统源码Java+IntelliJ+IDEA+MySQL一款先进的药物警戒系统

Java语言ADR药物不良反应系统源码JavaIntelliJIDEAMySQL一款先进的药物警戒系统源码 ADR药物不良反应监测系统是一个综合性的监测平台&#xff0c;旨在收集、报告、分析和评价药品在使用过程中可能出现的不良反应&#xff0c;以确保药品的安全性和有效性。 以下是对该系统的详细…

C#【进阶】俄罗斯方块

俄罗斯方块 文章目录 Test1_场景切换相关BeginScene.csBegionOrEndScene.csEndScene.csGame.csGameScene.csISceneUpdate.cs Test2_绘制对象基类和枚举信息DrawObject.csIDraw.csPosition.cs Test3_地图相关Map.cs Test4_坐标信息类BlockInfo.cs Test5_板砖工人类BlockWorker.…

04_前端三大件JS

文章目录 JavaScript1.JS的组成部分2.JS引入2.1 直接在head中通过一对script标签定义脚本代码2.2创建JS函数池文件&#xff0c;所有html文件共享调用 3.JS的数据类型和运算符4.分支结构5.循环结构6.JS函数的声明7.JS中自定义对象8.JS_JSON在客户端使用8.1JSON串格式8.2JSON在前…

Python与OpenCV:图像处理与计算机视觉实战指南

前言 OpenCV&#xff08;Open Source Computer Vision Library&#xff09;是一个开源的计算机视觉和机器学习软件库&#xff0c;它包含了数百种计算机视觉算法&#xff0c;包括图像处理、视频分析、物体检测、面部识别等。结合Python语言的强大功能&#xff0c;OpenCV可以用于…

java医院管理系统源码(springboot+vue+mysql)

风定落花生&#xff0c;歌声逐流水&#xff0c;大家好我是风歌&#xff0c;混迹在java圈的辛苦码农。今天要和大家聊的是一款基于springboot的医院管理系统。项目源码以及部署相关请联系风歌&#xff0c;文末附上联系信息 。 项目简介&#xff1a; 医院管理系统的主要使用者分…

【AJAX前端框架】Asynchronous Javascript And Xml

1 传统请求及缺点 传统的请求都有哪些&#xff1f; 直接在浏览器地址栏上输入URL。点击超链接提交form表单使用JS代码发送请求 window.open(url)document.location.href urlwindow.location.href url… 传统请求存在的问题 页面全部刷新导致了用户的体验较差。传统的请求导…

Android15 Beta更新速览

Android15 Beta更新速览 前台服务变更 前台服务使应用保持活动状态&#xff0c;以便它们可以执行关键且对用户可见的操作&#xff0c;通常以牺牲电池寿命为代价。在 Android 15 Beta 2 中&#xff0c;dataSync 和 mediaProcessing 前台服务类型现在具有约 6 小时的超时时间&a…

Python 植物大战僵尸

文章目录 效果图项目结构实现思路源代码 效果图 项目结构 实现思路 下面是代码的实现思路&#xff1a; 导入必要的库和模块&#xff1a;首先&#xff0c;我们导入了Python的os、time库以及pygame库&#xff0c;还有植物大战僵尸游戏中用到的各个植物和僵尸的类。 初始化游戏和…

AI写作工具的革命:AIGC如何提升内容生产效率

AIGC&#xff0c;即人工智能生成内容&#xff0c;是一种新兴的内容生产方式&#xff0c;它利用人工智能技术来自动生成文本、图像、音频、视频等多种形式的内容即进入实际应用层面。 所以AI不再是高深的、让人望尘莫及的算力算法&#xff0c;而是真实地贴近了我们的生活&#…

VectorDBBench在windows的调试

VectorDBBench在windows的调试 VectorDBBench是一款向量数据库基准测试工具&#xff0c;支持milvus、Zilliz Cloud、Elastic Search、Qdrant Cloud、Weaviate Cloud 、 PgVector、PgVectorRS等&#xff0c;可以测试其QPS、时延、recall。 VectorDBBench是一款使用python编写的…