大数据教程:Transformation和Action算子演示

大数据教程:Transformation和Action算子演示大数据教程:Transformation和Action算子演示

大家好,又见面了,我是你们的朋友全栈君。

大数据教程:Transformation和Action算子演示

一、Transformation算子演示

val conf = new SparkConf().setAppName(“Test”).setMaster(“local”)
val sc = new SparkContext(conf)

//通过并行化生成rdd

val 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 rdd7
println(rdd8.collect.toBuffer)

//intersection:求交集

val rdd9 = rdd6 intersection rdd7
println(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_2
println(rdd10_3.collect().toBuffer)

//左连接和右连接

//除基准值外是Option类型,因为可能存在空值所以使用Option

val rdd10_4 = rdd10_1 leftOuterJoin rdd10_2

//以左边为基准没有是null

val rdd10_5 = rdd10_1 rightOuterJoin rdd10_2

//以右边为基准没有是null

println(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_2
println(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.groupByKey
println(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_2
println(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)

转载于:https://juejin.im/post/5d07547d6fb9a07ef7107428

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/106770.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • webservice体系结构中包括_致命框架1第六关

    webservice体系结构中包括_致命框架1第六关   Web服务可以用来解决跨网络应用集合问题的开发模式,目的是保证不同平台的应用服务可以相互操作 JAX-WS实现WebServicepackagecom.service;importjavax.jws.WebService;/** WebService准备发布的接口* @WebService注解说明该类为Web服务发布类*/@WebServi…

  • integer constant is too large_unsigned int最大值

    integer constant is too large_unsigned int最大值用Integer存储sina的10位数字的uid,然后出现了input错,我以为Integer存10位数字没问题啊,确实能存10位,但最大值是2147483647,超过就不行了….

  • 关于Set集合去重[通俗易懂]

    关于Set集合去重[通俗易懂]上次说了List集合的去重,这次来说说Set集合的去重publicstaticvoidmain(String[]args){ HashSeths=newHashSet<>(); hs.add(“a”); hs.add(“b”); hs.add(“c”); hs.add(“a”); hs.add(“a”); for(Objectobj:…

  • linux中文镜像文件iso下载地址,linux系统镜像iso文件下载

    linux中文镜像文件iso下载地址,linux系统镜像iso文件下载Asp&period;NetCore–基于声明的授权翻译如下:当创建身份时,其可以被分配由可信方发布的一个或多个声明.索赔是名称值对,表示主题是什么,而不是主体可以做什么.例如,您可能有驾驶执照,由当地驾驶执照颁发.您的驾驶执照上有您的出生日期…优秀IT技术文章集&lpar;最新&rpar;&lpar;高质量&rpar;作者:赵磊博客:h…

  • softmax回归算法实验内容_最小二乘法回归模型

    softmax回归算法实验内容_最小二乘法回归模型简介在本节中,我们介绍Softmax回归模型,该模型是logistic回归模型在多分类问题上的推广,在多分类问题中,类标签  可以取两个以上的值。Softmax回归模型对于诸如MNIST手写数字分类等问题是很有用的,该问题的目的是辨识10个不同的单个数字。Softmax回归是有监督的,不过后面也会介绍它与深度学习/无监督学习方法的结合。(译者注:MNIST是一个手写数字识别库,由NYU的Y…

  • pytorch(8)– resnet101 迁移学习记录

    pytorch(8)– resnet101 迁移学习记录一、前言本篇记录使用pytorch官方resnet101实现迁移学习,迁移学习是当前深度学习领域的一系列通用的解决方案,而不是一个具体的算法模型。Pre-training+fine-tuning(预训练+调参)的迁移学习方式是现在深度学习中一个非常流行的迁移学习方式,有以下3步(1)把预训练模型当做特征提取器:TensorFlow或者Pytorch都有ImageNet上预训练好的模型,将最后一层全连接层(原始的是1000个类别或者更多)改成你自己的分类任务的种类进行输出,或…

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号