python - 在 pyspark 中使用嵌套元素从 RDD 获取平面 RDD
问题描述
我在 Pyspark 中有两个带有嵌套元素的 RDD,如下所示:
a = sc.parallelize(( (1,2), 3,(4,(6,7,(8,9,(11),10)),5,12)))
b = sc.parallelize(1,2,(3,4))
嵌套可以有任何深度。
我想合并它们,然后找到任何深度的最大元素,所以我尝试将其转换为 RDD,而没有像这样的嵌套值 (1,2,3,4,6,7,8,9,11,10, 5,12,1,2,3,4) 并获得最大值,使用其中任何一个(map、reduce、filter、flatmap、lamda 函数)。谁能告诉我如何转换或获得最大元素。
我提供了一个解决方案,但它仅适用于两个深度级别,例如
a = sc.parallelize(( (1,2), 3,(4,5)))
b = sc.parallelize((2,(4,6,7),8))
def maxReduce(tup):
return int(functools.reduce(lambda a,b : a if a>b else b, tup))
maxFunc = lambda x: maxReduce(x) if type(x) == tuple else x
a.union(b).map(lambda x: maxFunc(x)).reduce(lambda a,b : a if a>b else b)
上面的代码只适用于深度二,我需要为任何给定的深度工作(1,(2,3,(4,5,(6,(7,(8))))))
。
解决方案
听起来是递归函数的一个很好的用例:
from collections import Iterable
a = sc.parallelize(((1, 2), 3, (4, (6, 7, (8, 9, (11), 10)), 5, 12)))
b = sc.parallelize((1, 2, (3, 4)))
def maxIterOrNum(ele):
"""
this method finds the maximum value in an iterable otherwise return the value itself
:param ele: An iterable of numeric values or a numeric value
:return: a numeric value
"""
res = -float('inf')
if isinstance(ele, Iterable):
for x in ele:
res = max(res, maxIterOrNum(x))
return res
else:
return ele
a.union(b).reduce(lambda x, y: max(maxIterOrNum(x), maxIterOrNum(y)))
推荐阅读
- google-maps - 谷歌的查询和地方自动完成之间的区别?
- python - LookupError:未知编码:utf8r
- c# - 反序列化具有混合值 System.Text.JSON 的 JSON 数组
- javascript - 提交表单引导表单、innerHTML、警报
- amazon-web-services - AWS S3 Glacier 上传存档需要很长时间才能完成执行 - 检查状态或加快上传速度的方法?
- c# - 使用 CSharp 在 Kafka 中自定义序列化
- javascript - iframe 视频控件不可点击?
- html - 图像未调整大小
- php - jQuery 使用链接属性模拟在页面加载时单击链接
- html - CSS Navbar 项目不按顺序排列?