大数据分布式计算工具Spark数据计算实战讲解（map方法，flatmap方法，reducebykey方法）

��

数据计算

map方法

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

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

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

from pyspark import SparkConf, SparkContextimport osos.environ['pyspark_python'] = "D:/python/JIESHIQI/python.exe"#创建一个sparkconf类对象conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")#基于sparkconf类对象创建sparkcontext类对象sc = SparkContext(conf=conf)#准备一个rddrdd = sc.parallelize([1,2,3,4,5])#通过map方法将全部数据都乘以10def 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, SparkContextimport osos.environ['pyspark_python'] = "D:/python/JIESHIQI/python.exe"#创建一个sparkconf类对象conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")#基于sparkconf类对象创建sparkcontext类对象sc = SparkContext(conf=conf)#准备一个RDDrdd = 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, SparkContextimport osos.environ['pyspark_python'] = "D:/python/JIESHIQI/python.exe"#创建一个sparkconf类对象conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app")#基于sparkconf类对象创建sparkcontext类对象sc = SparkContext(conf=conf)#准备一个rddrdd = sc.parallelize([('男',99),('男',66),('女',33),('女',88)])#求男生和女生两个组的成绩之和rdd2 = rdd.reduceByKey(lambda a,b: a+b)print(rdd2.collect()) sc.stop()#[('女', 121), ('男', 165)]