2

这是我在 Coursera 上做的数据科学课程中的硬件任务的求助电话，因为我无法在 Coursera 论坛上获得任何建议。我已经编写了代码，但不幸的是输出没有返回预期的结果。这是手头的问题：

任务：将关系连接实现为 MapReduce 查询

输入（映射器）：

输入将是格式化为字符串列表的数据库记录。每个列表元素对应于其对应记录中的不同字段。每条记录中的第一项（索引 0）是一个字符串，用于标识该记录来自哪个表。该字段有两个可能的值：

  1. 'line_item' 表示该记录是一个行项目。2.'order'表示该记录是一个订单。

每条记录中的第二个元素（索引 1）是 order_id。LineItem 记录有 17 个元素，包括标识符字符串。订单记录有 10 个元素，包括标识符字符串。

输出（减速机）：

输出应该是一个连接的记录。

结果应该是一个长度为 27 的列表，其中包含来自订单记录的字段，后跟来自行项目记录的字段。每个列表元素都应该是一个字符串。

我的代码是：

import MapReduce
import sys

"""
Word Count Example in the Simple Python MapReduce Framework
"""

mr = MapReduce.MapReduce()

# =============================
# Do not modify above this line

record = open(sys.argv[1]) # this read input, given by instructor

def mapper(record):
key = record[1] # assign order_id from each record as key
value = list(record) # assign whole record as value for each key
mr.emit_intermediate(key, value) # emit key-value pairs

def reducer(key, value):
    new_dict = {} # create dict to keep track of records
    if not key in new_dict:
        new_dict[key] = value
    else:
        new_dict[key].extend(value)
    for key in new_dict:
        if len(new_dict[key]) == 27:
            mr.emit(new_dict[key])

# Do not modify below this line
# =============================
if __name__ == '__main__':
  inputdata = open(sys.argv[1])
  mr.execute(inputdata, mapper, reducer)

我收到的错误消息是“预期：31 条记录，得到 0”。

此外，预期的输出记录应该是这样的——只有一个列表，所有记录集中在一起，没有任何重复数据删除。

["order", "5", "44485", "F", "144659.20", "1994-07-30", "5-LOW", "Clerk#000000925", "0", "quickly. bold deposits sleep slyly. packages use slyly", "line_item", "5", "37531", "35", "3", "50", "73426.50", "0.08", "0.03", "A", "F", "1994-08-08", "1994-10-13", "1994-08-26", "DELIVER IN PERSON", "AIR", "eodolites. fluffily unusual"]

很抱歉问题很长，而且有点乱，但我希望答案对某人来说是显而易见的。

对我有用的类似代码：

def mapper(record):
    # key: document identifier
    # value: document contents
    friend = record[0]
    value = 1
    mydict = {}
    mr.emit_intermediate(friend, value)
    mydict[friend] = int(value)


def reducer(friend, value):
    # key: word
    # value: list of occurrence counts
    newdict = {}
    if not friend in newdict:
        newdict[friend] = value
    else:
    newdict[friend] = newdict[friend] + 1
    for friend in newdict:
    mr.emit((friend, (newdict[friend])))

谢谢！谢尔盖

4

4 回答 4

3

实际上你不必使用 new_dict。由于您必须打印“加入”并且您知道订单始终在值列表中的索引 0 中，并且列表的其余部分是line_item ，因此该代码应该这样做：

import MapReduce
import sys

"""
Word Count Example in the Simple Python MapReduce Framework
"""

mr = MapReduce.MapReduce()

# =============================
# Do not modify above this line

def mapper(record):
    key = record[1] # assign order_id from each record as key
    value = list(record) # assign whole record as value for each key
    mr.emit_intermediate(key, value) # emit key-value pairs

def reducer(key, value):
    for index in range (1, len(value)):
        mr.emit(value[0] + value[index])

# Do not modify below this line
# =============================
if __name__ == '__main__':
  inputdata = open(sys.argv[1])
  mr.execute(inputdata, mapper, reducer)
于 2014-08-05T16:44:01.310 回答
1

我看到这段代码有一些问题。首先是这一行：

record = open(sys.argv[1])

我觉得奇怪的是，这个record变量从未在代码的其他任何地方使用过。即使mapper函数定义如下：

def mapper(record):
    ...

...这 record是mapper功能的本地。它的范围与第一个不同record。无论传递给什么数据，mapper都会分配给它的本地并相应地使用，并且永远不会触及record分配给第一个的文件对象。record不过，我不认为这与错误有关。因为第一个record没有在其他任何地方使用，您可以非常安全地删除该行。

然后是reducer函数：

def reducer(key, value): # reducer should take 2 inputs according to the task
    if key in new_dict: # checking if key already added to dict
        new_dict[key].extend(list(value)) # if yes just append all records to the value
    new_dict[key] = list(value) # if not create new key and assign record to value
    for key in new_dict:
        if len(new_dict[key]) == 27: # checks to emit only records found in both tables
            mr.emit(new_dict[key])

您自己的评论为这里的问题提供了线索。首先，您说您正在检查密钥是否已经在字典中。如果是这样，只需将所有记录附加到该值。如果不是，则创建一个新键并将记录分配给该值。

问题在于与“如果不是”注释关联的行。如果第一次if测试失败时确实应该这样做，那么它应该以else一行开头：

    ...
    if key in new_dict: # checking if key already added to dict
        new_dict[key].extend(list(value)) # if yes just append all records to the value
    else:
        new_dict[key] = list(value) # if not create new key and assign record to value
    ...

您编写它的方式，即使该if测试成功并将数据附加到键的现有值，它也会立即踩下该更改。换句话说，该键的值不会增长。它总是代表最近提交的键值。

这是使用所有建议更改编辑的完整代码：

import MapReduce
import sys

"""
Word Count Example in the Simple Python MapReduce Framework
"""

mr = MapReduce.MapReduce()

# =============================
# Do not modify above this line

def mapper(record):
    key = record[1] # assign order_id from each record as key
    value = list(record) # assign whole record as value for each key
    mr.emit_intermediate(key, value) # emit key-value pairs

new_dict = {} # create dict to keep track of records

def reducer(key, value):
    if not key in new_dict:
        new_dict[key] = value
    else:
        new_dict[key].extend(value)
    for key in new_dict:
        if len(new_dict[key]) == 27:
            mr.emit(new_dict[key])

# Do not modify below this line
# =============================
if __name__ == '__main__':
  inputdata = open(sys.argv[1])
  mr.execute(inputdata, mapper, reducer)
于 2013-06-08T19:53:37.223 回答
0 
import MapReduce
import sys

mr = MapReduce.MapReduce()
new_dict = {}

def mapper(record):
   order_id = record[1]
    mr.emit_intermediate(order_id, list(record))


def reducer(key, list_of_lists):
    for lyst in list_of_lists:
    order_id = lyst[1]
    if not (order_id in new_dict):
    new_dict[order_id] = lyst
    else:
    new_dict[order_id].extend(lyst)
    mr.emit(new_dict[order_id])

if __name__ == '__main__':
    inputdata = open(sys.argv[1])
    #inputdata = open("records.json")  
    mr.execute(inputdata, mapper, reducer)
于 2016-03-18T07:28:35.977 回答
0

不知道它是否仍然相关，只是想指出，正如贾斯汀建议的那样，每次调用减速器时都会重置 new_dict 。一种方法是发出 2 个键值对。说你有 2 行 -

  1. order_id, line_item_id,Order_name

    1    ,      2      ,   'abc'

  2. line_item_id, line_item_location

    2        ,     'xyz'

我们希望输出为 -

 1,2,'abc','xyz'

从映射器发出 2 个键值对，连接列作为公共键 -

(2,[1,'abc']) and (2,['xyz'])

在减速器中，输入将是 -

(2,[[1,'abc'],['xyz']])

从那里开始尝试操作数据以获得所需的输出。

（供参考 - 我正在为我的 M/R 工作使用 python dumbo 框架）

于 2015-08-12T14:02:57.440 回答