赞
踩
1、RDD: 弹性的分布式数据集
弹性:RDD 中的数据即可以缓存在内存中, 也可以缓存在磁盘中, 也可以缓存在外部存储中
分布式:数据可以分布在多台服务器中,RDD中的分区来自于block块,而block块会来自不同的datanode
数据集:(1)RDD自身可以不存储数据的,只存放代码计算逻辑,触发作业执行的时候,数据会在RDD之间流动
(2)RDD 也可以缓存起来, 相当于存储具体数据
2、RDD 是一个分布式计算框架, 所以, 一定是要能够进行分区计算的, 只有分区了, 才能利用集群的并行计算能力
spark的运行过程中如果出现了相同的键被拉取到对应的分区,这个过程称之为shuffle。
注:只有 Key-Value
型的 RDD 才会有 Shuffle 操作,Spark的shuffle和mapreduce的shuffle原理是一样,都是要进行落盘。
-
- object Demo2Partition {
- def main(args: Array[String]): Unit = {
- //1、创建Spark环境
- //1.1 创建配置文件对象
- val conf: SparkConf = new SparkConf()
-
- //1.2 指定运行的模式(local Standalone Mesos YARN)
- conf.setMaster("local") //可以执行所运行需要核数资源local[2],不指定的话默认使用所有的资源执行程序
-
- //1.3 给spark作业起一个名字
- conf.setAppName("wc")
-
- //2、创建spark运行时的上下文对象
- //SparkContext是spark-core的入口组件, 是一个 Spark 程序的入口,主要作用是连接集群, 创建 RDD, 累加器, 广播变量等
- val sparkContext: SparkContext = new SparkContext(conf)
-
- //3、读取文件数据
- // val wordsLine: RDD[String] = sparkContext.textFile("spark/data/ws/*", minPartitions = 7)
- val wordsLine: RDD[String] = sparkContext.textFile("spark/data/ws/*")
- println(s"wordsLineRDD分区数是:${wordsLine.getNumPartitions}")
-
- //4、每一行根据|分隔符进行切分
- val words: RDD[String] = wordsLine.flatMap(_.split("\\|"))
- println(s"wordsRDD分区数是:${words.getNumPartitions}")
-
- val wordsTuple2: RDD[(String, Int)] = words.map((_, 1))
- println(s"wordsTuple2RDD分区数是:${wordsTuple2.getNumPartitions}")
-
- //产生shuffle的算子上可以单独设置分区数
- val wordsTuple2Group: RDD[(String, Iterable[(String, Int)])] = wordsTuple2.groupBy(_._1, 5)
- println(s"wordsTuple2GroupRDD分区数是:${wordsTuple2Group.getNumPartitions}")
-
- val wordCount: RDD[(String, Int)] = wordsTuple2Group.map((kv: (String, Iterable[(String, Int)])) => (kv._1, kv._2.size))
- println(s"wordCountRDD分区数是:${wordCount.getNumPartitions}")
-
- wordCount.saveAsTextFile("spark/data/word_count2")
-
- }
- }
SparkContext是spark功能的主要入口。其代表与spark集群的连接,能够用来在集群上创建RDD、累加器、广播变量。每个JVM里只能存在一个处于激活状态的SparkContext,在创建新的SparkContext之前必须调用stop()来关闭之前的SparkContext。
每一个Spark应用都是一个SparkContext实例,可以理解为一个SparkContext就是一个spark application的生命周期,一旦SparkContext创建之后,就可以用这个SparkContext来创建RDD、累加器、广播变量,并且可以通过SparkContext访问Spark的服务,运行任务。spark context设置内部服务,并建立与spark执行环境的连接。
sparkContext在Spark应用程序的执行过程中起着主导作用,它负责与程序和spark集群进行交互,包括申请集群资源、创建RDD、accumulators及广播变量等。
1、RDD是由一些分区构成的,读取文件时有多少个block块,RDD中就会有多少个分区 注:默认情况下,所有的RDD中的分区数是一样的,无论是shuffle之前还是shuffle之后的,在最开始加载数据的时候决定的
2、函数实际上是作用在RDD中的分区上的,一个分区是由一个task处理,有多少个分区,总共就有多少个task
注:函数在spark中称之为算子(转换transformation算子 RDD-->RDD,行动action算子 RDD->Other数据类型)
3、RDD之间存在一些依赖关系,后一个RDD中的数据是依赖与前一个RDD的计算结果,数据像水流一样在RDD之间流动
注:
3.1 RDD之间有两种依赖关系
a. 窄依赖 后一个RDD中分区数据对应前一个RDD中的一个分区数据 1对1的关系
b. 宽依赖 后一个RDD中分区数据来自前一个RDD中的多个分区数据 1对多的关系(shuffle)
3.2 因为有了依赖关系,将整个作业划分了一个一个stage阶段 sumNum(stage) = Num(宽依赖) + 1
3.3 窄依赖的分区数是不可以改变,取决于第一个RDD分区数,宽依赖可以在产生shuffle的算子上设置分区数
4、分区类的算子只能作用在键值对格式的RDD上,groupByKey、reduceByKey
5、spark为task计算提供了精确的计算位置,移动计算而不移动数据
RDD 中的算子从功能上分为两大类
Transformation(转换) :它会在一个已经存在的 RDD 上创建一个新的 RDD, 将旧的 RDD 的数据转换为另外一种形式后放入新的 RDD
Action(行动): 执行各个分区的计算任务, 将的到的结果返回到 Driver 中
注意:RDD具有懒执行的特点,一个spark作业,由action算子来触发执行的,若没有action算子,整个作业不执行
将rdd中的数据,一条一条的取出来传入到map函数中,map会返回一个新的rdd,map不会改变总数据条数
- object Demo3Map {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("map算子演示")
- val context = new SparkContext(conf)
- //====================================================
-
- val studentRDD: RDD[String] = context.textFile("spark/data/students.csv")
-
- /**
- * map算子:将rdd中的数据,一条一条的取出来传入到map函数中,map会返回一个新的rdd,map不会改变总数据条数
- */
- val splitRDD: RDD[List[String]] = studentRDD.map((s: String) => {
- println("============好好学习================")
- s.split(",").toList
- })
-
- // splitRDD.foreach(println) //foreach是action算子
- }
- }
filter: 过滤,将RDD中的数据一条一条取出传递给filter后面的函数,如果函数的结果是true,该条数据就保留,否则丢弃
- import org.apache.spark.{SparkConf, SparkContext}
- import org.apache.spark.rdd.RDD
-
- object Demo4Filter {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("map算子演示")
- val context = new SparkContext(conf)
- //====================================================
- val studentRDD: RDD[String] = context.textFile("spark/data/students.csv")
-
- /**
- * filter: 过滤,将RDD中的数据一条一条取出传递给filter后面的函数,如果函数的结果是true,该条数据就保留,否则丢弃
- *
- * filter一般情况下会减少数据的条数
- */
- val filterRDD: RDD[String] = studentRDD.filter((s: String) => {
- val strings: Array[String] = s.split(",")
- "男".equals(strings(3))
- })
-
-
- filterRDD.foreach(println)
-
- }
- }
flatMap算子:将RDD中的数据一条一条的取出传递给后面的函数,函数的返回值必须是一个集合。最后会将集合展开构成一个新的RDD
- import org.apache.spark.rdd.RDD
- import org.apache.spark.{SparkConf, SparkContext}
-
- object Demo5flatMap {
- def main(args: Array[String]): Unit = {
-
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("flatMap算子演示")
- val context = new SparkContext(conf)
- //====================================================
- val linesRDD: RDD[String] = context.textFile("spark/data/words.txt")
-
- /**
- * flatMap算子:将RDD中的数据一条一条的取出传递给后面的函数,函数的返回值必须是一个集合。最后会将集合展开构成一个新的RDD
- */
- val wordsRDD: RDD[String] = linesRDD.flatMap((line: String) => line.split("\\|"))
-
- wordsRDD.foreach(println)
- }
- }
从前一个RDD的数据中抽样一部分数据,抽取的比例不是正好对应的,在抽取的比例上下浮动 比如1000条抽取10% 抽取的结果在100条左右
- import org.apache.spark.rdd.RDD
- import org.apache.spark.{SparkConf, SparkContext}
-
- object Demo6sample {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("flatMap算子演示")
- val context = new SparkContext(conf)
- //====================================================
- val studentRDD: RDD[String] = context.textFile("spark/data/students.csv")
-
- /**
- * sample算子:从前一个RDD的数据中抽样一部分数据
- *
- * 抽取的比例不是正好对应的,在抽取的比例上下浮动 比如1000条抽取10% 抽取的结果在100条左右
- */
- val sampleRDD: RDD[String] = studentRDD.sample(withReplacement = true, 0.1)
- sampleRDD.foreach(println)
- }
-
- }
groupBy:按照指定的字段进行分组,返回的是一个键是分组字段,值是一个存放原本数据的迭代器的键值对 返回的是kv格式的RDD
- object Demo7GroupBy {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("groupBy算子演示")
- val context = new SparkContext(conf)
- //====================================================
- val studentRDD: RDD[String] = context.textFile("spark/data/students.csv")
-
- val splitRDD: RDD[Array[String]] = studentRDD.map((s: String) => s.split(","))
-
- //需求:求出每个班级平均年龄
- //使用模式匹配的方式取出班级和年龄
- val clazzWithAgeRDD: RDD[(String, Int)] = splitRDD.map {
- case Array(_, _, age: String, _, clazz: String) => (clazz, age.toInt)
- }
-
- /**
- * groupBy:按照指定的字段进行分组,返回的是一个键是分组字段,值是一个存放原本数据的迭代器的键值对 返回的是kv格式的RDD
- *
- * key: 是分组字段
- * value: 是spark中的迭代器
- * 迭代器中的数据,不是完全被加载到内存中计算,迭代器只能迭代一次
- *
- * groupBy会产生shuffle
- */
- //按照班级进行分组
- //val stringToStudents: Map[String, List[Student]] = stuList.groupBy((s: Student) => s.clazz)
- val kvRDD: RDD[(String, Iterable[(String, Int)])] = clazzWithAgeRDD.groupBy(_._1)
- val clazzAvgAgeRDD: RDD[(String, Double)] = kvRDD.map {
- case (clazz: String, itr: Iterable[(String, Int)]) =>
- //CompactBuffer((理科二班,21), (理科二班,23), (理科二班,21), (理科二班,23), (理科二班,21), (理科二班,21), (理科二班,24))
- //CompactBuffer(21,23,21,23,21,21,24)
- val allAge: Iterable[Int] = itr.map((kv: (String, Int)) => kv._2)
- val avgAge: Double = allAge.sum.toDouble / allAge.size
- (clazz, avgAge)
- }
-
- clazzAvgAgeRDD.foreach(println)
-
- while (true){
-
- }
-
- }
-
- }
groupByKey: 按照键进行分组,将value值构成迭代器返回,在spark中看到RDD[(xx, xxx)] 这样的RDD就是kv键值对类型的RDD,只有kv类型键值对RDD才可以调用groupByKey算子。
- import org.apache.spark.{SparkConf, SparkContext}
- import org.apache.spark.rdd.RDD
-
- object Demo8GroupByKey {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("groupByKey算子演示")
- val context = new SparkContext(conf)
- //====================================================
- val studentRDD: RDD[String] = context.textFile("spark/data/students.csv")
-
- val splitRDD: RDD[Array[String]] = studentRDD.map((s: String) => s.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()
- val clazzAvgAgeRDD: RDD[(String, Double)] = kvRDD.map {
- case (clazz: String, ageItr: Iterable[Int]) =>
- (clazz, ageItr.sum.toDouble / ageItr.size)
- }
- clazzAvgAgeRDD.foreach(println)
-
- while (true){
-
- }
-
- /**
- * groupBy与groupByKey的区别(spark的面试题)
- * 1、代码上的区别:任意一个RDD都可以调用groupBy算子,只有kv类型的RDD才可以调用groupByKey
- * 2、groupByKey之后产生的RDD的结构比较简单,方便后续处理
- * 3、groupByKey的性能更好,执行速度更快,因为groupByKey相比较与groupBy算子来说,shuffle所需要的数据量较少
- */
- }
- }
按照键key对value值直接进行聚合,需要传入聚合的方式,reduceByKey算子也是只有kv类型的RDD才能调用。
- import org.apache.spark.{SparkConf, SparkContext}
- import org.apache.spark.rdd.RDD
-
- object Demo9ReduceByKey {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("reduceByKey算子演示")
- val context = new SparkContext(conf)
- //====================================================
- val studentRDD: RDD[String] = context.textFile("spark/data/students.csv")
-
- val splitRDD: RDD[Array[String]] = studentRDD.map((s: String) => s.split(","))
-
- //求每个班级的人数
- val clazzKVRDD: RDD[(String, Int)] = splitRDD.map {
- case Array(_, _, _, _, clazz: String) => (clazz, 1)
- }
-
- /**
- * 利用groupByKey实现
- */
- // val kvRDD: RDD[(String, Iterable[Int])] = clazzKVRDD.groupByKey()
- // val clazzAvgAgeRDD: RDD[(String, Double)] = kvRDD.map {
- // case (clazz: String, n: Iterable[Int]) =>
- // (clazz, n.sum)
- // }
- // clazzAvgAgeRDD.foreach(println)
-
- /**
- * 利用reduceByKey实现:按照键key对value值直接进行聚合,需要传入聚合的方式
- * reduceByKey算子也是只有kv类型的RDD才能调用
- *
- *
- */
- val countRDD: RDD[(String, Int)] = clazzKVRDD.reduceByKey((x: Int, y: Int) => x + y)
- countRDD.foreach(println)
-
-
-
- // clazzKVRDD.groupByKey()
- // .map(kv=>(kv._1,kv._2.sum))
- // .foreach(println)
-
- while (true){
-
- }
-
- /**
- * reduceByKey与groupByKey的区别
- * 1、reduceByKey比groupByKey在map端多了一个预聚合的操作,预聚合之后的shuffle数据量肯定是要少很多的,性能上比groupByKey要好
- * 2、从灵活角度来看,reduceByKey并没有groupByKey灵活
- * 比如reduceByKey无法做方差,groupByKey后续可以完成
- *
- */
-
-
- }
- }
union:上下合并两个RDD,前提是两个RDD中的数据类型要一致,合并后不会对结果进行去重。这里的合并只是逻辑层面上的合并,物理层面其实是没有合并
- import org.apache.spark.rdd.RDD
- import org.apache.spark.{SparkConf, SparkContext}
-
- object Demo10Union {
- def main(args: Array[String]): Unit = {
- 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") // 1
- val w2RDD: RDD[String] = context.textFile("spark/data/ws/w2.txt") // 1
-
- /**
- * union:上下合并两个RDD,前提是两个RDD中的数据类型要一致,合并后不会对结果进行去重
- *
- * 注:这里的合并只是逻辑层面上的合并,物理层面其实是没有合并
- */
-
- val unionRDD: RDD[String] = w1RDD.union(w2RDD)
- println(unionRDD.getNumPartitions) // 2
-
- unionRDD.foreach(println)
- while (true){
-
- }
-
- }
- }
- import org.apache.spark.rdd.RDD
- import org.apache.spark.{SparkConf, SparkContext}
- /**
- * 内连接:join
- * 左连接:leftJoin
- * 右连接:rightJoin
- * 全连接:fullJoin
- */
-
- object Demo11Join {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("Join算子演示")
- val context = new SparkContext(conf)
- //====================================================
-
- //两个kv类型的RDD之间的关联
- //通过scala中的集合构建RDD
- val rdd1: RDD[(String, String)] = context.parallelize(
- List(
- ("1001", "尚平"),
- ("1002", "丁义杰"),
- ("1003", "徐昊宇"),
- ("1004", "包旭"),
- ("1005", "朱大牛"),
- ("1006","汪权")
- )
- )
-
- val rdd2: RDD[(String, String)] = context.parallelize(
- List(
- ("1001", "崩坏"),
- ("1002", "原神"),
- ("1003", "王者"),
- ("1004", "修仙"),
- ("1005", "学习"),
- ("1007", "敲代码")
- )
- )
-
-
-
- //内连接
- // val innerJoinRDD: RDD[(String, (String, String))] = rdd1.join(rdd2)
- // //加工一下RDD
- // val innerJoinRDD2: RDD[(String, String, String)] = innerJoinRDD.map {
- // case (id: String, (name: String, like: String)) => (id, name, like)
- // }
- // innerJoinRDD2.foreach(println)
-
- //左连接
- val leftJoinRDD: RDD[(String, (String, Option[String]))] = rdd1.leftOuterJoin(rdd2)
- //加工一下RDD
- val leftJoinRDD2: RDD[(String, String, String)] = leftJoinRDD.map {
- case (id: String, (name: String, Some(like))) => (id, name, like)
- case (id: String, (name: String, None)) => (id, name, "无爱好")
- }
- leftJoinRDD2.foreach(println)
-
- println("=================================")
-
- //右连接与左连接相差不多,不在赘述
-
- //全连接
- val fullJoinRDD: RDD[(String, (Option[String], Option[String]))] = rdd1.fullOuterJoin(rdd2)
- //加工一下RDD
- val fullJoinRDD2: RDD[(String, String, String)] = fullJoinRDD.map {
- case (id: String, (Some(name), Some(like))) => (id, name, like)
- case (id: String, (Some(name), None)) => (id, name, "无爱好")
- case (id: String, (None, Some(like))) => (id, "无姓名", like)
- }
- fullJoinRDD2.foreach(println)
-
- }
-
- }
- import org.apache.spark.rdd.RDD
- import org.apache.spark.{SparkConf, SparkContext}
-
- object Demo12Student {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("sortby算子演示")
- val context = new SparkContext(conf)
- //====================================================
- //需求:统计总分年级排名前10的学生的各科分数
- //读取分数文件数据
- val scoreRDD: RDD[(String, String, String)] = context.textFile("spark/data/score.txt") // 读取数据文件
- .map((s: String) => s.split(",")) // 切分数据
- .filter((arr: Array[String]) => arr.length == 3) // 过滤掉脏数据
- .map {
- //整理数据,进行模式匹配取出数据
- case Array(sid: String, subject_id: String, score: String) => (sid, subject_id, score)
- }
-
- //计算每个学生的总分
- val sumScoreWithSidRDD: RDD[(String, Int)] = scoreRDD.map {
- case (sid: String, _: String, score: String) => (sid, score.toInt)
- }.reduceByKey((x: Int, y: Int) => x + y)
-
-
- //按照总分排序
- val sumScoreTop10: Array[(String, Int)] = sumScoreWithSidRDD.sortBy(-_._2).take(10)
-
- //取出前10的学生学号
- val ids: Array[String] = sumScoreTop10.map(_._1)
-
- //取出每个学生各科分数
- val top10StuScore: RDD[(String, String, String)] = scoreRDD.filter {
- case (id: String, _, _) => ids.contains(id)
- }
-
- top10StuScore.foreach(println)
-
- }
- }
也是作用在kv类型的RDD上,主要的作用键不变,处理值
- import org.apache.spark.{SparkConf, SparkContext}
- import org.apache.spark.rdd.RDD
-
- object Demo13MapValues {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("Join算子演示")
- val context = new SparkContext(conf)
- //====================================================
- //需求:统计总分年级排名前10的学生的各科分数
- //读取分数文件数据
- val scoreRDD: RDD[(String, String, String)] = context.textFile("spark/data/score.txt") // 读取数据文件
- .map((s: String) => s.split(",")) // 切分数据
- .filter((arr: Array[String]) => arr.length == 3) // 过滤掉脏数据
- .map {
- //整理数据,进行模式匹配取出数据
- case Array(sid: String, subject_id: String, score: String) => (sid, subject_id, score)
- }
-
- //计算每个学生的总分
- val sumScoreWithSidRDD: RDD[(String, Int)] = scoreRDD.map {
- case (sid: String, _: String, score: String) => (sid, score.toInt)
- }.reduceByKey((x: Int, y: Int) => x + y)
-
- /**
- * mapValues算子:也是作用在kv类型的RDD上
- * 主要的作用键不变,处理值
- */
- val resRDD: RDD[(String, Int)] = sumScoreWithSidRDD.mapValues(_ + 1000)
- resRDD.foreach(println)
-
- //等同于
- val res2RDD: RDD[(String, Int)] = sumScoreWithSidRDD.map((kv: (String, Int)) => (kv._1, kv._2 + 1000))
- }
- }
mapPartition: 主要作用是一次处理一个分区的数据,将一个分区的数据一个一个传给后面的函数进行处理
- import org.apache.spark.{SparkConf, SparkContext}
- import org.apache.spark.rdd.RDD
-
- object Demo14mapPartition {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("mapPartition算子演示")
- val context = new SparkContext(conf)
- //====================================================
- //需求:统计总分年级排名前10的学生的各科分数
- //读取分数文件数据
- val scoreRDD: RDD[String] = context.textFile("spark/data/ws/*") // 读取数据文件
-
- println(scoreRDD.getNumPartitions)
-
- /**
- * mapPartition: 主要作用是一次处理一个分区的数据,将一个分区的数据一个一个传给后面的函数进行处理
- *
- * 迭代器中存放的是一个分区的数据
- */
- // val mapPartitionRDD: RDD[String] = scoreRDD.mapPartitions((itr: Iterator[String]) => {
- //
- // println(s"====================当前处理的分区====================")
- // //这里写的逻辑是作用在一个分区上的所有数据
- // val words: Iterator[String] = itr.flatMap(_.split("\\|"))
- // words
- // })
-
- // mapPartitionRDD.foreach(println)
-
- scoreRDD.mapPartitionsWithIndex{
- case (index:Int,itr: Iterator[String]) =>
- println(s"当前所处理的分区编号是:${index}")
- itr.flatMap(_.split("\\|"))
- }.foreach(println)
-
- }
-
- }
以数组的形式返回数据集中所有元素
- import org.apache.spark.rdd.RDD
- import org.apache.spark.{SparkConf, SparkContext}
-
- object Demo15Actions {
- def main(args: Array[String]): Unit = {
- val conf = new SparkConf()
- conf.setMaster("local")
- conf.setAppName("Action算子演示")
- val context = new SparkContext(conf)
- //====================================================
-
- val studentRDD: RDD[String] = context.textFile("spark/data/students.csv")
-
- /**
- * 转换算子:transformation 将一个RDD转换成另一个RDD,转换算子是懒执行的,需要一个action算子触发执行
- *
- * 行动算子(操作算子):action算子,触发任务执行。一个action算子就会触发一次任务执行
- */
- println("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$")
- val studentsRDD: RDD[(String, String, String, String, String)] = studentRDD.map(_.split(","))
- .map {
- case Array(id: String, name: String, age: String, gender: String, clazz: String) =>
- println("**************************** ^_^ ********************************")
- (id, name, age, gender, clazz)
- }
- println("$$$$$$$$$$$$$$$$$$$$$$***__***$$$$$$$$$$$$$$$$$$$$$$$$$")
-
- // foreach其实就是一个action算子
- // studentsRDD.foreach(println)
- // println("="*100)
- // studentsRDD.foreach(println)
-
- // while (true){
- //
- // }
-
- /**
- * collect()行动算子 主要作用是将RDD转成scala中的数据结构
- *
- */
- val tuples: Array[(String, String, String, String, String)] = studentsRDD.collect()
-
- }
-
- }
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。