Spark第三篇：pyspark下的key-value函数

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了Spark第三篇：pyspark下的key-value函数，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含3191字，纯文字阅读大概需要5分钟。

内容图文

partitionBy

目的：对源数据进行重新分区，

def partitionBy(self, numPartitions, partitionFunc=portable_hash):

其中只需要指定numPartitions就可以了

reduceByKey

目的：可以将数据按照相同的Key对Value进行聚合
示例代码：

from pyspark import SparkConf, SparkContext

# 创建local表示只用单线程，loacal[*]表示用电脑全部的cpu核
conf = SparkConf().setMaster("local[*]").setAppName("lichao-wordcount")
sc = SparkContext(conf=conf)
# 输入的数据
data = [("a",3),("a",5),("a",1),("b",4)]
rdd = sc.parallelize(data)
#聚合函数，进行两两聚合
redrdd = rdd.reduceByKey(lambda x,y:x+y)

resultColl = redrdd.collect()
for line in resultColl:
    print(line)

结果：

('b', 4)
('a', 9)

groupByKey

目的：将数据源的数据根据key对value进行分组
示例代码：

from pyspark import SparkConf, SparkContext

# 创建local表示只用单线程，loacal[*]表示用电脑全部的cpu核
conf = SparkConf().setMaster("local[*]").setAppName("lichao-wordcount")
sc = SparkContext(conf=conf)
# 输入的数据
data = [("a",3),("a",5),("a",1),("b",4)]
rdd = sc.parallelize(data)
#分组函数，keu相同的进行两两分组
grprdd = rdd.groupByKey()

resultColl = grprdd.collect()
for line in resultColl:
    print(line)

结果：

('b', <pyspark.resultiterable.ResultIterable object at 0x00000269CE39EAC8>)
('a', <pyspark.resultiterable.ResultIterable object at 0x00000269CE39EA88>)

分析：
从shuffle的角度：reduceByKey和groupByKey都存在shuffle的操作，但是reduceByKey可以在shuffle前对分区内相同key的数据进行预聚合（combine）功能，这样会减少落盘的数据量，而groupByKey只是进行分组，不存在数据量减少的问题，reduceByKey性能比较高。
从功能的角度：reduceByKey其实包含分组和聚合的功能。GroupByKey只能分组，不能聚合，所以在分组聚合的场合下，推荐使用reduceByKey，如果仅仅是分组而不需要聚合。那么还是只能使用groupByKey

aggregateByKey

目的：将数据根据不同的规则进行分区内计算和分区间计算，其中还需要指定初始参数zeroValue，

    def aggregateByKey(self, zeroValue, seqFunc, combFunc, numPartitions=None,
                       partitionFunc=portable_hash):

对比：既可以指定分区内的计算规则，也可以指定分区间的计算规则，groupByKey中，分区内的分组操作，分区间也是分组操作，可以说是一种特例。

示例代码：分区内进行分类，分区间进行聚合

# groupBYkey和rreducebykey分区内的逻辑和分区间逻辑一定是一样的，但是如果分区内要进行最大值比较，分区间要进行相加聚合，那就不行
# agg可以完成分区内和分区间计算逻辑的不同的效果

from pyspark import SparkConf, SparkContext
import math

# 创建local表示只用单线程，loacal[*]表示用电脑全部的cpu核
conf = SparkConf().setMaster("local[*]").setAppName("lichao-wordcount")
sc = SparkContext(conf=conf)
# 输入的数据
data = [("a",3),("a",5),("a",1),("b",4)]
rdd = sc.parallelize(data)
#agg分组聚合函数
#参数1：初始值，因为最初的时候只读进来一个数字，如果没有初始值，那就不能进行最大值最小值的比较
#参数2：分区内计算逻辑的函数
#参数3：分区间计算逻辑的函数
aggrdd = rdd.aggregateByKey(0,lambda x,y:max(x,y),lambda x,y:x+y)

resultColl = aggrdd.collect()
for line in resultColl:
    print(line)
    

结果：

('b', 4)
('a', 9)

foldByKey

目的：指定规则对数据源的分区内和分区间进行操作，但是分区内和分区间必须一样
函数说明：用法和aggregateByKey类似

    def foldByKey(self, zeroValue, func, numPartitions=None, partitionFunc=portable_hash):

sortByKey

目的：一句键值对数据进行排序

 def sortByKey(self, ascending=True, numPartitions=None, keyfunc=lambda x: x):

内容总结

以上是互联网集市为您收集整理的Spark第三篇：pyspark下的key-value函数全部内容，希望文章能够帮你解决Spark第三篇：pyspark下的key-value函数所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---