如何在 pyspark 中的 map/lambda 中使用 map/lambda？答案

【问题标题】：How to put use a map/lambda inside of a map/lambda in pyspark?如何在 pyspark 中的 map/lambda 中使用 map/lambda？
【发布时间】：2015-12-12 03:05:39
【问题描述】：

我有一个这样的列表：

b = [['r','w'],['n','finished']]

我希望能够对每个 list 中的每个 element 进行操作。

我可以在 python 本地执行此操作：

result = b.map(lambda aList: \
           map(lambda aString: \
                              '' if aString.strip().lower() in [' finish', 'finished', 'terminate', 'done'] else aString,\
                              aList))

但是，Spark 在序列化内部 map 时遇到了问题：

 File "/<path>/python/pyspark/worker.py", line 88, in main
12/11/2015 18:24:49 [launcher]      command = pickleSer._read_with_length(infile)
12/11/2015 18:24:49 [launcher]    File "//<path>/spark/python/pyspark/serializers.py", line 156, in _read_with_length
12/11/2015 18:24:49 [launcher]      return self.loads(obj)
12/11/2015 18:24:49 [launcher]    File "//<path>//python/pyspark/serializers.py", line 405, in loads
12/11/2015 18:24:49 [launcher]      return cPickle.loads(obj)
12/11/2015 18:24:49 [launcher]  AttributeError: 'module' object has no attribute 'map'

如何解决这个问题，使用内部地图或完成相同的事情？

【问题讨论】：

标签： python lambda apache-spark pyspark

【解决方案1】：

一种处理方法：

to_replace = ['finish', 'finished', 'terminate', 'done'] 

rdd = sc.parallelize([['r','w'],['n','finished']])
rdd.map(lambda xs: ['' if  x.strip().lower() in to_replace else x for x in xs])

一般来说，如果您发现自己在考虑嵌套函数，这是一个好兆头，您应该使用普通函数而不是 lambda 表达式。

【讨论】：

“正常功能”是指built-ins?
定义一个函数并在 map 中使用它，而不是使用 lambda 表达式。比如def myfunc(x):和rdd.map(myfunc)

【解决方案2】：

或者使用@zero323 的模板，如果您使用的是Python 2.x，您可以使用map 而不是for，但这是python 问题而不是pyspark 问题，并且效果是一样的。

to_replace = ['finish', 'finished', 'terminate', 'done'] 

rdd = sc.parallelize([['r','w'],['n','finished']])
rdd.map(
    lambda xs: map(lambda x: "" if x.strip().lower() in to_replace else x, xs)
)

但是，如果to_replace 列表真的很大，你应该使用broadcast variable。

【讨论】：