在分布式环境中计算精确中位数需要一些努力,因此假设您想要类似于RDD中所有值的平方。我们称这个方法为squares
,并假设它应按以下方式工作:
assert rdd.squares().collect() == rdd.map(lambda x: x * x).collect()
1. 修改 pyspark.RDD
定义:
from pyspark import RDD
def squares(self):
return self.map(lambda x: x * x)
RDD.squares = squares
rdd = sc.parallelize([1, 2, 3])
assert rdd.squares().collect() == [1, 4, 9]
注意:如果您修改类定义,每个实例都将可以访问squares
。
2. 创建RDD子类:
class RDDWithSquares(RDD):
def squares(self):
return self.map(lambda x: x * x)
rdd = sc.parallelize([1, 2, 3])
rdd.__class__ = RDDWithSquares
给一个类分配一个类别是一种不良的hack,因此在实践中应该以适当的方式创建RDD(例如,请参见context.parallelize的实现)。
3. 将方法添加到实例
import types
rdd = sc.parallelize([1, 2, 3])
rdd.squares = types.MethodType(squares, rdd)
免责声明
首先,我没有测试过这些内容足够长的时间来确保没有隐藏的问题。
此外,我认为这并不值得大惊小怪。如果没有静态类型检查,很难找到任何好处,您可以使用函数、柯里化和pipes
以更清晰的方式获得类似的结果。
from toolz import pipe
pipe(
sc.parallelize([1, 2, 3]),
squares,
lambda rdd: rdd.collect())
RDD [Any]
。因此,您有责任仅调用适用的方法。类似于sc.parallelize(range(3)).groupByKey()
这样的东西显然是没有意义的,并且在执行转换时会失败,但在类型级别上没有任何问题。 - zero323