spark-MLlib_计算机软件及应用_IT计算机_专业资料

《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()