1. textFile() 读取数据源，返回一个string类型的RDD，
举例：
val conf = new SparkConf().setAppName(“appname”).setMaster("local")
val sparkContext = new SparkContext(conf)
myRdd = sparkContext.textFile() ；
2. map() 遍历每条数据（或进行相应处理）,主要用于数据过滤和处理
myRdd = sparkContext.textFile() ；
val resultRdd = myRdd .map(x = (x.split("\t")(1), x.split("\t")(2) + "\t" + x.split("\t")(3)))
3.keys和values 返回RDD的key的集合和value的集合
val keysResult = resultRdd.keys.collect
val valuesResult = resultRdd.values
4. collect() 返回一个数组
val array = esultRdd.collect()
注意：collect()函数会把所有数据读取到JVM的内存中,故只有在数据量比较小是才使用collect()函数,否则会导致JVM内存溢出,切记!
5.count() 返回RDD的element的个数
val count = resultRdd.count()
6.take(Int n) 返回前n个element
7.takeOrdered(n, [ordering]) 根据指定排序(升序,降序或自定义)方式,返回前n个element
8.reduceByKey(fun) 指定函数,将相同的key进行聚合计算
val wordCount = reduceByKey((x,y) => {x + y }) // 返回key相等，对应的value的和
val wordCount = reduceByKey((x,y) => { // 返回key相等的value的最大值
if(x > y) {
x
} else {
y
}
})
9.groupBy(fun) 根据指定字段进行分组，key相等的分为一个
10.groupByKey(fun) 函数和reduce很相识，但是实现原理有细微差别具体如下：
如下基于常用的wordCount为例，详解reduceByKey和groupByKey的过程，代码如下：
val words = Array("one", "two", "two", "three", "three", "three") val wordPairsRDD = sc.parallelize(words).map(word => (word, 1)) val wordCountsWithReduce = wordPairsRDD .reduceByKey(_ + _) .collect() val wordCountsWithGroup = wordPairsRDD .groupByKey() .map(t => (t._1, t._2.sum)) .collect()
a.groupByKey的shuffle过程：
groupByKey的过程
b.reduceByKey函数实在同一台机器上的数据在shuffle之前，对相同的key的数据进行聚合到一起，减少了无所谓的网路传输，然后再将不同分区的数据聚合，最终得到结果；
reduceByKey的过程
因此要避免使用groupByKey
11.filter(fun) 传入函数fun,过滤集合中的数据
12.sortBy() 根据传入的字段进行排序
13. sortByKey() 根据key进行排序,可指定升序或者是降序;
14.partitionBy() 根据传入的字段进行数据分区
15.mapPartitions() 对每个分区的数据进行处理,和map很像(map是针对所有数据处理)
若一个有10个分区的rdd有10000个元素，使用map方法，map中的输入函数会被调用10000次；而使用mapPartitions方法的话，其输入函数 会只会被调用10次，每个分区调用1次。其中返回的数据需要转换为iterator
16.aggregate() 待完成
17.x.join(y) 类似于SQL的inner join操作，只返回x.key和y.key配对成功的，过滤关联不到的数据
val resultRdd = rddX.join(rddY).map(x => {
val rddXKey = x._1._1 //rddX的map中的key
val rddXValue = x._1._2 //rddX的map中的value
val rddYKey = x._2._1 //rddY的map中的key
val rddYValue = x._2._2 // rddY的map中的value
})
-
x.leftOuterJoin(y) 左外链接 似于SQL中的左外关联left outer join,返回所有x的值，针对没有对应的y的值返回null
-
x.rightOuterJoin(y) 左外链接 似于SQL中的左外关联left outer join,返回所有y的值，针对没有对应的x的值返回null
网友评论