用法

当RDD分区内和分区间算法不同时,使用aggregate aggregateByKey等2个方法;
但是当分区内和分区间算法相同时,便可以简化使用fold foldByKey这2个方法;

使用示例

# -*- coding: utf-8 -*-
"""
(C) rgc
All rights reserved
create time '2021/5/26 19:37'Usage:
fold 用法
foldByKey 用法
"""
# 构建spark
from pyspark.conf import SparkConf
from pyspark.context import SparkContextconf = SparkConf()
# 使用本地模式;且 executor设置为1个方便debug
conf.setMaster('local[1]').setAppName('rgc')
sc = SparkContext(conf=conf)rdd = sc.parallelize([2, 1, 3, 4, 4], 1)def seq_func(x: tuple, y: int or float) -> tuple:"""计算函数:param x: 指  zeroValue 参数 也就是(2,0):param y: 指 具体的元素:return: 元素的累加,元素的个数"""return x[0] + y, x[1] + 1def comb_func(x: tuple, y: tuple) -> tuple:"""计算函数:param x: 第一个元素 类型 元祖:param y: 第二个元素 类型 元祖:return:"""return x[0] + y[0], x[1] + y[1]# fold:rdd分区内和分区间使用相同的函数进行聚合计算;
# 注意: 初始值 会参与分区内和分区间的计算;
# 所以 每个分区内的一次初始值(5) + 分区内的元素(2+1+3+4+4) + 分区间的一次初始值(5) = 24
rdd_res = rdd.map(lambda x: (x, 1)).fold((5, 0), comb_func)
print(rdd_res)  # (24,5)# foldByKey:rdd分区内 所有元素先根据key进行分组,然后分区内和分区间使用 相同的函数 对每个key对应的values 进行聚合计算;
# 注意: 初始值 只参与分区内的计算,不参与分区间计算
rdd_res = rdd.map(lambda x: (x, 1)).foldByKey((1, 0), seq_func)
print(rdd_res.collect())  # [(2, (2, 1)), (1, (2, 1)), (3, (2, 1)), (4, (3, 2))]

注意

fold方法和aggregate通过查看源码发现其会调用collect()方法;而collect()方法会将Executor端每个分区所有执行的结果收集到Driver端,然后在Driver端进行分区间操作;也就是Driver端如果内存不够时会报错;

PySpark fold foldByKey用法相关推荐

pyspark dataframe基本用法
pyspark dataframe基本用法 #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Created on F ...
PySpark reduce reduceByKey用法
用法 reduce:对rdd内部元素进行迭代操作 reduce方法分区内和分区间调用相同的用户给定的函数; 先在每个分区内执行完用户给定的函数后,将每个分区的结果通过collect()方法统计到 ...
when-otherwise for pyspark用法
pyspark when otherwise用法描述:根据某一列的值,修改另一列的值 1.第一种思路,生成临时表,统计数量,用withcolumn 修改数据 2.第二种思路,调用pandas API ...
kotlin之plus、copyOf、reverse、forEach、filter、map、reduce、fold等函数解释和使用
kotlin之::函数调用.plus(增加元素).copyOf(复制数组).reverse(翻转数组).forEach(遍历数组).filter(过滤数组).map函数操作及扩展.reduce函数.f ...
boost::hana::fold用法的测试程序
boost::hana::fold用法的测试程序实现功能 C++实现代码实现功能 boost::hana::fold用法的测试程序 C++实现代码 #include <boost/hana/ ...
boost::fusion::fold用法的测试程序
boost::fusion::fold用法的测试程序实现功能 C++实现代码实现功能 boost::fusion::fold用法的测试程序 C++实现代码 #include <boost/f ...
Python大数据处理扩展库pySpark用法精要
Spark是一个开源的.通用的并行计算与分布式计算框架,其活跃度在Apache基金会所有开源项目中排第三位,最大特点是基于内存计算,适合迭代计算,兼容多种应用场景,同时还兼容Hadoop生态系统中的组 ...
pyspark及Spark报错问题汇总及某些函数用法。
此贴,主要记录本人在工作中遇到的某些报错问题,并提出自己的解决办法. 1. spark = SparkSession.builder() TypeError: 'Builder' object is ...
Pytorch nn.Fold()的简单理解与用法
官方文档:https://pytorch.org/docs/stable/generated/torch.nn.Fold.html 这个东西基本上就是绑定Unfold使用的.实际上,在没有overla ...
PySpark与GraphFrames的安装与使用
PySpark环境搭建配置hadoop spark访问本地文件并执行运算时,可能会遇到权限问题或是dll错误.这是因为spark需要使用到Hadoop的winutils和hadoop.dll,首先我 ...

PySpark fold foldByKey用法

用法

使用示例

注意

PySpark fold foldByKey用法相关推荐

最新文章

热门文章