Spark_4：Spark常见操作函数

以{1, 2, 3, 3}的RDD为例：

函数名	目的	示例	结果
map()	将函数应用于 RDD 中的每个元素，将返回值构成新的 RDD	rdd.map(x => x + 1)	{2, 3, 4, 4}
flatMap()	将函数应用于 RDD 中的每个元素，将返回的迭代器的所有内容构成新的 RDD。通常用来切分单词	rdd.flatMap(x => x.to(3))	{1, 2, 3, 2, 3, 3, 3}
filter()	返回一个由通过传给 filter() 的函数的元素组成的 RDD	rdd.filter(x => x != 1)	{2, 3, 3}
distinct()	去重	rdd.distinct()	{1, 2, 3}
sample(withReplacement, fraction, [seed])	对 RDD 采样，以及是否替换	rdd.sample(false, 0.5)	非确定的

表3-3：对数据分别为{1, 2, 3}和{3, 4, 5}的RDD进行针对两个RDD的转化操作

函数名	目的	示例	结果
union()	生成一个包含两个 RDD 中所有元素的 RDD	rdd.union(other)	{1, 2, 3, 3, 4, 5}
intersection()	求两个 RDD 共同的元素的 RDD	rdd.intersection(other)	{3}
subtract()	移除一个 RDD 中的内容（例如移除训练数据）	rdd.subtract(other)	{1, 2}
cartesian()	与另一个 RDD 的笛卡儿积	rdd.cartesian(other)	{(1, 3), (1, 4), ...(3, 5)}

表3-4：对一个数据为{1, 2, 3, 3}的RDD进行基本的RDD行动操作

函数名	目的	示例	结果
collect()	返回 RDD 中的所有元素	rdd.collect()	{1, 2, 3, 3}
count()	RDD 中的元素个数	rdd.count()	4
countByValue()	各元素在 RDD 中出现的次数	rdd.countByValue()	{(1, 1),(2, 1),(3, 2)}
take(num)	从 RDD 中返回 num 个元素	rdd.take(2)	{1, 2}
top(num)	从 RDD 中 返回最前面的 num 个元素	rdd.top(2)	{3, 3}
takeOrdered(num)(ordering)	从 RDD 中按照提供的顺序返回最前面的 num 个元素	rdd.takeOrdered(2)(myOrdering)	{3, 3}
takeSample(withReplacement, num, [seed])	从 RDD 中返回任意一些元素	rdd.takeSample(false, 1)	非确定的
reduce(func)	并行整合 RDD 中 所有数据（例如 sum）	rdd.reduce((x, y) => x + y)	9
fold(zero)(func)	和 reduce() 一样，但是需要提供初始值	rdd.fold(0)((x, y) => x + y)	9
aggregate(zeroValue)(seqOp, combOp)	和 reduce() 相似，但是通常返回不同类型的函数	rdd.aggregate((0, 0)) ((x, y) => (x._1 + y, x._2 + 1), (x, y) => (x._1 + y._1, x._2 + y._2))复制代码	(9,4)
foreach(func)	对 RDD 中的每个元素使用给定的函数	rdd.foreach(func)	无

 

1、rdd.repartition()：对RDD重新分区

2、map() 和 flatMap()：

map(): 将函数应用于 RDD 中的每个元素，将返回值构成新的 RDD.

flatMap()：将函数应用于 RDD 中的每个元素，将返回的迭代器的所有内容构成新的 RDD。通常用来切分单词

3、distinct()：RDD去重

4、reduceByKey(add)：只针对key重复的二元组元素。接收一个函数，将重复key的value按函数聚合

from __future__ import print_function
from pyspark import SparkContext
from pyspark import SparkConf
 
conf = SparkConf().setMaster("local").setAppName("chapter4")
sc = SparkContext(conf=conf)
num = sc.parallelize([1, 4, 2, 3, 4, 4, 2, 4])
pairs = num.map(lambda x: (x, 1))  # 针对每个数进行计数1的操作，
print(pairs.collect())  #输出为列表，每个列表中都是一个二元组，key为每个数，value都为1
a = pairs.reduceByKey(lambda x, y: x+y+1) # 
b = pairs.reduceByKey(lambda x, y: x+y+2)
c = pairs.reduceByKey(lambda x, y: x+y+3)
print(a.collect())
print(b.collect())
print(c.collect())

# 输出

[(1, 1), (4, 1), (2, 1), (3, 1), (4, 1), (4, 1), (2, 1), (4, 1)]

[(1, 1), (2, 3), (3, 1), (4, 7)]

[(1, 1), (2, 4), (3, 1), (4, 10)]

[(1, 1), (2, 5), (3, 1), (4, 13)]

 

# 解释：

reduceByKey的函数是针对具有相同键的二元组。在这里例子中，key=1 和key=3都分别只有一个value =1，即他们的键只有1个，所以他们并不执行匿名函数lambda，所以输出都是（1，1）（3，1）。

而key=2的二元组对应了两个（2，1）（2，1），value1=1，value2=1，key=4的二元组有4个：（4，1）（4，1）（4，1）（4，1），value1=1，value2=1，value3=1，value4=1。这两个key就会执行lambda函数。

对于a来说，执行x+y+1，这个语句怎么理解呢？ --》 对于每个相同的key，其value相加并加1，那么key=2: 就有value1+value2+1 = 1+1+1=3。key=4: 就有value1+value2+1 +value3+1+value4+1= 1+1+1+1+1+1+1=7；

对于b来说，执行x+y+2 --》 对于每个相同的key，其value相加并加2，那么key=2: 就有value1+value2+2 = 1+1+2=5。key=4: 就有value1+value2+2 +value3+2+value4+2= 1+1+2+1+2+1+2=10；

对于c来说，执行x+y+3  --》 对于每个相同的key，其value相加并加3，那么key=2: 就有value1+value2+1 = 1+1+1=3。key=4: 就有value1+value2+3+value3+3+value4+3= 1+1+3+1+3+1+3=13；

 

综上 reduceByKey是针对每个具有相同键的值的操作，依次操作。