首页 > 代码库 > spark 教程三 spark Map filter flatMap union distinct intersection操作
spark 教程三 spark Map filter flatMap union distinct intersection操作
RDD的创建 spark 所有的操作都围绕着弹性分布式数据集（RDD）进行，这是一个有容错机制的并可以被并行操作的元素集合，具有只读、分区、容错、高效、无需物化、可以缓存、RDD依赖等特征
RDD的创建基础RDD
1.并行集合（Parallelized Collections）:接收一个已经存在的Scala集合，然后进行各种并行运算
var sc=new SparkContext(conf) var rdd=sc.parallelize(Array(2,4,9,3,5,7,8,1,6)); rdd.foreach(println)
2.hadoop 数据集（Hadoop Datasets）:在一个文件的每条记录上运行函数只要文件系统是HDFS，或者hadoop支持的任意存档存储系统即可
val file=sc.textFile("hdfs://hostname:9090/path/somefile.txt")Map的操作
map(func) 返回一个新的RDD ，由每一个元素经过func后转换组成
//设置conf master =local 为本地模式， appNamm 是应用程序的名称 var conf=new SparkConf().setMaster("local[1]").setAppName("RDD") var sc=new SparkContext(conf) // 加载一个数组，然后给没一个元素乘以2， 并且输出 var rdd=sc.parallelize(Array(2,4,9,3,5,7,8,1,6)).map(_*2).collect().foreach(println)filter(func) 返回一个新的RDD，由func函数计算后返回为true 的元素组成
//返回字母为b的元素 var strrdd=sc.parallelize(Array("a","b","c","d")).filter(_=="b").foreach(println)flatMap(func) 类似于Map 但是每一个输入元素被映射成0个或多个输出元素，因此func 应该返回一个序列，而不是单一元素
//把每一个元素乘以2然后在压平，输出 var list=sc.parallelize(List(List(1,6),List(2,7),List(3,8),List(4,9),List(5,10))) list.flatMap(_.map(_*2)).foreach(println)
下面用flatMap做一个wordcount的例子，新建一个文件建立了一下几个字段
/读取本地文件然后 。用split函数 把元素拆分成单一的元素，然后再用个map 给没个元素设置一个1， 在用reduceByKey把相同key的数量相加 var txtrdd=sc.textFile("d:/517/wc.txt").flatMap(_.split(" ")).map(x=>(x,1)).reduceByKey(_+_).foreach(println) result:(tome,1)(tom,1)(xiaoli,1)(lusi,1)(hello,7)(java,1)(jame,2)distinct(numTasks) 返回一个数据集中不包含重复元素的新数据集 （去重）
var la=sc.parallelize(List(("A",1),("A",6),("B",2),("B",7),("A",1))) var distinct=la.distinct()result:(A,1)(A,6)(B,7)(B,2)union(otherDataset) 返回一个新的数据集，新数据集由源数据集和参数数据集共同构成
var lb=sc.parallelize(List(("A",1),("A",6),("B",2),("B",7),("C",3))) var la=sc.parallelize(List(("A",1),("A",6),("D",2),("E",7),("A",1))) var distinct=la.union(lb) distinct.foreach(println result: (A,1) (A,6) (D,2) (E,7) (A,1) (A,1) (A,6) (B,2) (B,7) (C,3)
intersection(otherDataset) 返回一个新的数据集，新数据集由源数据集和参数数据集的交集构成
var lb=sc.parallelize(List(("A",1),("A",6),("B",2),("B",7),("C",3))) var la=sc.parallelize(List(("A",1),("A",6),("D",2),("E",7),("A",1))) var distinct=la.intersection(lb) distinct.foreach(println) result: (A,1) (A,6)
spark 教程三 spark Map filter flatMap union distinct intersection操作
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。