如何在Hadoop Streaming中使用Python实现计数器

Question

我是hadopp streaming的新手，我的reduce代码中有几个过滤条件，我想知道有多少记录通过了这些条件。我得知可以通过编写自定义计数器来实现此功能。请问有人可以指导我如何编写自定义计数器吗？

在mapper代码中，我输出了三列，比如说 a,b,c，键是a，值是列表，类似于 [b,c]，为了举一个mapper代码的例子，它看起来像 ['I' ^ ['C','P']]

以下是我的reduce代码。

labels = ["a","b"]
for line in sys.stdin:
    l = line.strip().split("^")
    key = l[0]
    value = l[1]
    record = [key] + value
    records.append(record)
df = pd.DataFrame.from_records(records,columns=labels)
df = df((df['a'] == 'I') & (df['b'] == 'C'))

我想知道在reducer层面，df包含多少条记录。

谢谢。

- subro

Answer 1

您可以简单地将内容打印到stderr：

print >> sys.stderr, "reporter:counter: CUSTOM, NbRecords,1"

这会将计数器组“CUSTOM”中的计数器“NbRecords”增加1

- user1151446

Answer 2

如果使用mrjob，

class MRCountingJob(MRJob):

    def mapper(self, _, value):
        self.increment_counter('group', 'counter_name', 1)
        yield _, value

如果使用基本的hadoop streaming API（使用Python），

sys.stderr.write("reporter:counter:group,counter_name,1\n")

其中，例如group可以是"My Mapper"、"My Reducer"或"My FooBar"，计数器可以是num_calls，通常值始终为1，因为这些将由框架求和。（使用stderr.write时，请不要忘记换行符\n）

- michael

---

如何访问计数器？比如说，如果脚本A运行了Hadoop Map Reduce命令，然后运行了Mapper.py和Reducer.py，那么脚本A有没有一种简单的方法来访问计数器/查看输出呢？ - Kent Wong