欢迎来到天天文库
浏览记录
ID:41568145
大小:84.21 KB
页数:3页
时间:2019-08-27
《spark-MLlib_计算机软件及应用_IT计算机_专业资料》由会员上传分享,免费在线阅读,更多相关内容在工程资料-天天文库。
1、SparkMlliblearningRDD(ResilientDistributeDatabases弹性分布式数据集):存储在不同节点计算机中的数据集。弹性指数据的存储方式和容错性强,可以是基于内存或者是磁盘,提供不同的持久化和运行方式,如果一个节点发生错误,RDD在不同节点重试。Transformation操作:mapjilter,groupByJoinAction操作:count,reduce,collectRDD所有操作都是Lazy加载模式,是一种程序优化的特殊形式,运行编译的过程中不会立即到计算机中计算
2、最终结果,而是记住所有操作和步骤,只有使用Action方法才会进行计算Actionresult丿RDDAPI讲解l.aggregate方法对给定的数据集进行方法的设定defaggregate[U:ClassTag](zeroValue:U)(seqop:(U,T)=>U/combOp:(U/U)=>U):UU为数据类型,可以传入任意数据类型数据,seqop给定的计算方法要和U类型匹配,combOp是合并方法将zeroValue值和seqop值进行合并代码示例importorg.apache.spark.{Spa
3、rkContext,SparkConf}objecttestRDDMethod{defmain(args:Array[String]){valconf=newSparkConf()〃创建上下文环境.setMasterC'local")〃设置本地化处理.setAppNameCtestRODMethod")//设置应用名称valsc=newSparkContext(conf)〃创建上下文环境实例valarr=sc.parallelize(Array(l/2,3/4,5,6))〃输入数组数据集valresult=ar
4、r.aggregate(O)(math.ma)<(_/_),_+_)〃使用aggregate方法println(result)}}输出结果:6parallelize方法将内存中数据读入到spark中,作为整体数据集,math.max比较数据集中数据的大小,是对传递的两个值进行求和,Aggregate对第一个方法结果和空值计算。2.parallelize是SparkContext中的方法defparallelize(T:ClassTag)(seq:Seq[T],numSlices:lnt=defaultParal
5、lelism):RDD[T]第一个测试是数据,第二个表示将数据分布在多少个数据节点中存放,默认为1importorg.apache.spark.{SparkContext,SparkConf}objecttestRDDMethod2{defmain(args:Array[String]){valconf=newSparkConf()〃设置上下文环境.setMasterf'local")〃设置本地化处理・setAppNarne(utestRDDMethod2")〃设置应用名称valsc=newSparkConte
6、xt(conf)〃实例化上下文环境valarr=sc.parallelized(Array(l/2/3/4/5/6)/2)//Array(l/2/3)+Array(4/5/6)valresult=arr.aggregate(O)(math.max(_/_),_+_)pritnln(result)}}Result:3+6=9Aggregate用于字符串importorg.apache.spark.{SparkContext,SparkConf}objecttestRDDMethod2{defmain(args:A
7、rray[String]){valconf=newSparkConf()〃设置上下文环境.setMasterC'local")〃设置本地化处理・setAppNarne(utestRDDMethod2")〃设置应用名称valsc=newSparkContext(conf)〃实例化上下文环境〃输入数组数据集,第二个参数表示数据值分布在多少个数据节点中存放,默认为1valarr二sc・parallelize(Array「abcTbTcTdeTfg”))〃创建数据集〃调用aggregate方法valresult=arr
8、.aggregate(,'")((value/vord)=>value+word/_+_)pritnln(result)}}3.cache()操作将数据保存到内存中importorg.apache.spark{SparkConfzSparkContext}objectCacheTestfdefmain(args;Array[String]){valconf=newSparkConf()
此文档下载收益归作者所有