深圳大数据培训：Transformation算子演示

深圳大数据培训：Transformation算子演示
val conf = new SparkConf().setAppName(“Test”).setMaster(“local”)

  val sc = new SparkContext(conf)//通过并行化生成rddval rdd = sc.parallelize(List(5,6,4,7,3,8,2,9,10))//map:对rdd里面每一个元乘以2然后排序val rdd2: RDD[Int] = rdd.map(_ * 2)//collect以数组的形式返回数据集的所有元素(是Action算子)println(rdd2.collect().toBuffer)//filter:该RDD由经过func函数计算后返回值为true的输入元素组成val rdd3: RDD[Int] = rdd2.filter(_ > 10)println(rdd3.collect().toBuffer)val rdd4 = sc.parallelize(Array("a b c","b c d"))//flatMap:将rdd4中的元素进行切分后压平val rdd5: RDD[String] = rdd4.flatMap(_.split(" "))println(rdd5.collect().toBuffer)//假如: List(List(" a,b" ,"b c"),List("e c"," i o"))//压平 flatMap(_.flatMap(_.split(" ")))//sample随机抽样//withReplacement表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样//fraction抽样比例例如30% 即0.3 但是这个值是一个浮动的值不准确//seed用于指定随机数生成器种子 默认参数不传val rdd5_1 = sc.parallelize(1 to 10)val sample = rdd.sample(false,0.5)println(sample.collect().toBuffer)//union:求并集val rdd6 = sc.parallelize(List(5,6,7,8))val rdd7 = sc.parallelize(List(1,2,5,6))val rdd8 = rdd6 union rdd7println(rdd8.collect.toBuffer)//intersection:求交集val rdd9 = rdd6 intersection rdd7println(rdd9.collect.toBuffer)//distinct:去重出重复println(rdd8.distinct.collect.toBuffer)//join相同的key会被合并val rdd10_1 = sc.parallelize(List(("tom",1),("jerry" ,3),("kitty",2)))val rdd10_2 = sc.parallelize(List(("jerry" ,2),("tom",2),("dog",10)))val rdd10_3 = rdd10_1 join rdd10_2println(rdd10_3.collect().toBuffer)//左连接和右连接//除基准值外是Option类型,因为可能存在空值所以使用Optionval rdd10_4 = rdd10_1 leftOuterJoin rdd10_2 //以左边为基准没有是nullval rdd10_5 = rdd10_1 rightOuterJoin rdd10_2 //以右边为基准没有是nullprintln(rdd10_4.collect().toList)println(rdd10_5.collect().toBuffer)val rdd11_1 = sc.parallelize(List(("tom",1),("jerry" ,3),("kitty",2)))val rdd11_2 = sc.parallelize(List(("jerry" ,2),("tom",2),("dog",10)))//笛卡尔积val rdd11_3 = rdd11_1 cartesian rdd11_2println(rdd11_3.collect.toBuffer)

//根据传入的参数进行分组

val rdd11_5_1 = rdd11_4.groupBy(_._1)println(rdd11_5_1.collect().toList)//按照相同key进行分组,并且可以制定分区val rdd11_5_2 = rdd11_4.groupByKeyprintln(rdd11_5_2.collect().toList)//根据相同key进行分组[分组的话需要二元组]//cogroup 和 groupBykey的区别//cogroup不需要对数据先进行合并就以进行分组 得到的结果是 同一个key 和不同数据集中的数据集合//groupByKey是需要先进行合并然后在根据相同key进行分组val rdd11_6: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd11_1 cogroup rdd11_2println(rdd11_6)

Action算子演示
val conf = new SparkConf().setAppName(“Test”).setMaster("local[]")
val sc = new SparkContext(conf)
/ Action 算子*/
//集合函数
val rdd1 = sc.parallelize(List(2,1,3,6,5),2)
val rdd1_1 = rdd1.reduce(+)
println(rdd1_1)
//以数组的形式返回数据集的所有元素
println(rdd1.collect().toBuffer)
//返回RDD的元素个数
println(rdd1.count())
//取出对应数量的值默认降序, 若输入0 会返回一个空数组
println(rdd1.top(3).toBuffer)
//顺序取出对应数量的值
println(rdd1.take(3).toBuffer)
//顺序取出对应数量的值默认生序
println(rdd1.takeOrdered(3).toBuffer)
//获取第一个值等价于 take(1)
println(rdd1.first())
//将处理过后的数据写成文件(存储在HDFS或本地文件系统)
//rdd1.saveAsTextFile(“dir/file1”)
//统计key的个数并生成map k是key名 v是key的个数
val rdd2 = sc.parallelize(List((“key1”,2),(“key2”,1),(“key3”,3),(“key4”,6),(“key5”,5)),2)
val rdd2_1: collection.Map[String, Long] = rdd2.countByKey()
println(rdd2_1)
//遍历数据
rdd1.foreach(x => println(x))

/*其他算子*/
//统计value的个数 但是会将集合中的一个元素看做是一个vluae
val value: collection.Map[(String, Int), Long] = rdd2.countByValue
println(value)
//filterByRange:对RDD中的元素进行过滤,返回指定范围内的数据
val rdd3 = sc.parallelize(List(("e",5),("c",3),("d",4),("c",2),("a",1)))
val rdd3_1: RDD[(String, Int)] = rdd3.filterByRange("c","e")//包括开始和结束的
println(rdd3_1.collect.toList)
//flatMapValues对参数进行扁平化操作,是value的值
val rdd3_2 = sc.parallelize(List(("a","1 2"),("b","3 4")))
println( rdd3_2.flatMapValues(_.split(" ")).collect.toList)
//foreachPartition 循环的是分区数据
// foreachPartiton一般应用于数据的持久化,存入数据库,可以进行分区的数据存储
val rdd4 = sc.parallelize(List(1,2,3,4,5,6,7,8,9),3)
rdd4.foreachPartition(x => println(x.reduce(_+_)))
//keyBy 以传入的函数返回值作为key ,RDD中的元素为value 新的元组
val rdd5 = sc.parallelize(List("dog","cat","pig","wolf","bee"),3)
val rdd5_1: RDD[(Int, String)] = rdd5.keyBy(_.length)
println(rdd5_1.collect.toList)
//keys获取所有的key  values 获取所有的values
println(rdd5_1.keys.collect.toList)
println(rdd5_1.values.collect.toList)
//collectAsMap  将需要的二元组转换成Map
val map: collection.Map[String, Int] = rdd2.collectAsMap()
println(map)

深圳大数据培训：Transformation算子演示相关推荐

深圳大数据培训：HDFS基本概念
深圳大数据培训:HDFS基本概念 1.1 HDFS的介绍源自于Google的GFS论文发表于2003年10月 HDFS是GFS克隆版 ,HDFS的全称是Hadoop Distributed Fil ...
深圳大数据培训：大数据开发之掌握Hive的静态分区与动态分区
深圳大数据培训:大数据开发之掌握Hive的静态分区与动态分区分区是hive存放数据的一种方式.将列值作为目录来存放数据,就是一个分区. 这样查询时使用分区列进行过滤,只需根据列值直接扫描对应目录下的 ...
深圳大数据培训技术分享：Hadoop集群同步
深圳大数据培训技术分享:Hadoop集群同步分享--是技术突飞猛进的很好体验!在千锋学习大数据技术,开始学会了分享,班里五十个人,每个人就能得到49份不同技术探讨.每次到分享的时刻,总会收获不同的想 ...
深圳大数据培训：好程序员大数据学习路线之hive 存储格式
深圳大数据培训:好程序员大数据学习路线之hive 存储格式好程序员大数据学习路线之hive存储格式,hive的存储格式通常是三种:textfile . sequencefile . rcfile . ...
深圳大数据培训：大数据技术可以解决哪些问题？
深圳大数据培训:大数据技术可以解决哪些问题? 去年,大数据专业成为很多高校爆款专业,比如复旦,中南等.为什么如此多的大学开设此专业?学习它之后能解决哪些问题?以后它会应用在哪些工作岗位上? 一. 为什 ...
2018大数据培训学习路线图（详细完整版）
2018大数据培训学习路线全课程目录+学习线路详解(详细完整版) 第一阶段:大数据基础Java语言基础阶段 1.1:Java开发介绍 1.1.1 Java的发展历史 1.1.2 Java的应用领域 1 ...
大数据培训Spark 高频面试考点分享
1.Spark 如何保证宕机迅速恢复? 适当增加 spark standby master 编写 shell 脚本,定期检测 master 状态,出现宕机后对 master 进行重启操作 2. Spa ...
大数据培训ClickHouse表引擎
表引擎表引擎(即表的类型)决定了: 1)数据的存储方式和位置,写到哪里以及从哪里读取数据 2)支持哪些查询以及如何支持. 3)并发数据访问. 4)索引的使用(如果存在). 5)是否可以执行多线程请求 ...
大数据培训：Spark性能调优与参数配置
Spark性能调优-基础篇众所周知,正确的参数配置对提升Spark的使用效率具有极大助力,帮助相关数据开发.分析人员更高效地使用Spark进行离线批处理和SQL报表分析等作业. 推荐参数配置模板如下 ...

深圳大数据培训：Transformation算子演示

深圳大数据培训：Transformation算子演示相关推荐

最新文章

热门文章