我有一个映射器方法:
def mapper(value):
...
for key, value in some_list:
yield key, value
实际上,我需要的与普通的字数示例相去不远。我已经有了工作脚本,但前提是映射器方法看起来像这样:
def mapper(value):
...
return key, value
它的调用如下所示:
sc.textFile(sys.argv[2], 1).map(mapper).reduceByKey(reducer).collect()
我花了 2 个小时尝试编写支持映射器生成器的代码。但做不到。我什至同意只返回一个列表:
def mapper(value):
...
result_list = []
for key, value in some_list:
result_list.append( key, value )
return result_list
这里: https://groups.google.com/forum/#!searchin/spark-users/flatmap$20multiple/spark-users/1WqVhRBaJsU/-D5QRbenlUgJ我发现我应该使用 flatMap,但它没有解决问题 - 我的化简器然后开始获取输入,例如 (key1, value1, key2, value2, value3, ...) - 但它应该是 [(key1, value1), (key2, value2, value3)...]。换句话说,reducer开始只取单个部分,并且不知道它是一个值还是一个键,如果值 - 它属于哪个键。
那么如何使用返回迭代器或列表的映射器呢?
谢谢!
如果你想要一个返回多个输出的映射函数,可以使用flatMap
。
传递给flatMap
的函数可以返回一个可迭代对象:
>>> words = sc.textFile("README.md")
>>> def mapper(line):
... return ((word, 1) for word in line.split())
...
>>> words.flatMap(mapper).take(4)
[(u'#', 1), (u'Apache', 1), (u'Spark', 1), (u'Lightning-Fast', 1)]
>>> counts = words.flatMap(mapper).reduceByKey(lambda x, y: x + y)
>>> counts.take(5)
[(u'all', 1), (u'help', 1), (u'webpage', 1), (u'when', 1), (u'Hadoop', 12)]
它也可以是生成器函数:
>>> words = sc.textFile("README.md")
>>> def mapper(line):
... for word in line.split():
... yield (word, 1)
...
>>> words.flatMap(mapper).take(4)
[(u'#', 1), (u'Apache', 1), (u'Spark', 1), (u'Lightning-Fast', 1)]
>>> counts = words.flatMap(mapper).reduceByKey(lambda x, y: x + y)
>>> counts.take(5)
[(u'all', 1), (u'help', 1), (u'webpage', 1), (u'when', 1), (u'Hadoop', 12)]
您提到您尝试过flatMap
但它将所有内容扁平化为列表[key, value, key, value, ...]
而不是键值对[(key, value), (key, value)...]
列表。 我怀疑这是您的地图功能中的问题。 如果您仍然遇到此问题,您可以发布更完整版本的地图功能吗?