数据计算

map方法

PySpark的数据计算，都是基于RDD对象来进行的，那么如何进行呢？

自然是依赖，RDD对象内置丰富的：成员方法（算子）

功能：map算子，是将rdd的数据一条条处理（处理的逻辑基于map算子中接收的处理函数），返回新的rdd

 

from pyspark import SparkConf, SparkContext
import os
os.environ['pyspark_python'] = "D:/python/JIESHIQI/python.exe"#创建一个sparkconf类对象
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")#基于sparkconf类对象创建sparkcontext类对象
sc = SparkContext(conf=conf)#准备一个rdd
rdd = sc.parallelize([1,2,3,4,5])
#通过map方法将全部数据都乘以10
def func(data):return data * 10rdd2 = rdd.map(func)
#rdd2 = rdd.map(lambda x: x * 10)print(rdd2.collect())sc.stop()
#[10, 20, 30, 40, 50]#链式调用
rdd3 = rdd2.map(lambda x: x+5)
rdd2 = rdd.map(lambda x: x * 10).map(lambda x: x + 5)
#[15, 25, 35, 45, 55]

map算子（成员方法）

• 接受一个处理函数，可用lambda表达式快速编写

• 对RDD内的元素逐个处理，并返回一个新的RDD

链式调用

• 对于返回值是新RDD的算子，可以通过链式调用的方式多次调用算子。

flatmap算子

功能：对rdd执行map操作，然后进行解除嵌套操作

from pyspark import SparkConf, SparkContext
import os
os.environ['pyspark_python'] = "D:/python/JIESHIQI/python.exe"#创建一个sparkconf类对象
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")#基于sparkconf类对象创建sparkcontext类对象
sc = SparkContext(conf=conf)#准备一个RDD
rdd = sc.parallelize(["itcast sxx wyx", "qeq apple banana"])
#需求，将rdd数据里面的一个个单词提取处理
rdd2 = rdd.flatMap(lambda x: x.split(" "))
print(rdd2.collect())
sc.stop()
#['itcast', 'sxx', 'wyx', 'qeq', 'apple', 'banana']

reducebykey方法

功能：针对kv型rdd，自动按照key分组，然后根据你提供的聚合逻辑，完成组内数据的聚合操作

from pyspark import SparkConf, SparkContext
import os
os.environ['pyspark_python'] = "D:/python/JIESHIQI/python.exe"#创建一个sparkconf类对象
conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")#基于sparkconf类对象创建sparkcontext类对象
sc = SparkContext(conf=conf)#准备一个rdd
rdd = sc.parallelize([('男',99),('男',66),('女',33),('女',88)])
#求男生和女生两个组的成绩之和
rdd2 = rdd.reduceByKey(lambda a,b: a+b)
print(rdd2.collect()) 
sc.stop()
#[('女', 121), ('男', 165)]