从 IPython 笔记本运行 MRJob答案

【问题标题】：Run MRJob from IPython notebook从 IPython 笔记本运行 MRJob
【发布时间】：2014-09-02 06:40:23
【问题描述】：

我正在尝试从 IPython 笔记本运行 mrjob 示例

from mrjob.job import MRJob


class MRWordFrequencyCount(MRJob):

def mapper(self, _, line):
    yield "chars", len(line)
    yield "words", len(line.split())
    yield "lines", 1

def reducer(self, key, values):
    yield key, sum(values)

然后用代码运行它

mr_job = MRWordFrequencyCount(args=["testfile.txt"])
with mr_job.make_runner() as runner:
    runner.run()
    for line in runner.stream_output():
        key, value = mr_job.parse_output_line(line)
        print key, value

并得到错误：

TypeError: <module '__main__' (built-in)> is a built-in class

有没有办法从 IPython notebook 运行 mrjob？

【问题讨论】：

标签： python mapreduce ipython-notebook mrjob

【解决方案1】：

我还没有找到“完美的方法”，但您可以做的一件事是创建一个笔记本单元格，使用 %%file 魔法，将单元格内容写入文件：

%%file wordcount.py
from mrjob.job import MRJob

class MRWordFrequencyCount(MRJob):

    def mapper(self, _, line):
        yield "chars", len(line)
        yield "words", len(line.split())
        yield "lines", 1

    def reducer(self, key, values):
        yield key, sum(values)

然后让mrjob 在稍后的单元格中运行该文件：

import wordcount
reload(wordcount)

mr_job = wordcount.MRWordFrequencyCount(args=['example.txt'])
with mr_job.make_runner() as runner:
    runner.run()
    for line in runner.stream_output():
        key, value = mr_job.parse_output_line(line)
        print key, value

请注意，我调用了我的文件wordcount.py，并且我从wordcount 模块导入了类MRWordFrequencyCount——文件名和模块必须匹配。 Python 还缓存导入的模块，当您更改 wordcount.py-file 时，iPython 不会重新加载模块，而是使用旧的缓存模块。这就是我将reload() 调用放在那里的原因。

参考：https://groups.google.com/d/msg/mrjob/CfdAgcEaC-I/8XfJPXCjTvQJ

更新（更短）
对于较短的第二个笔记本单元，您可以通过从笔记本中调用 shell 来运行 mrjob

! python mrjob.py shakespeare.txt

参考：http://jupyter.cs.brynmawr.edu/hub/dblank/public/Jupyter%20Magics.ipynb

【讨论】：

【解决方案2】：

我怀疑这是由于 MRJob 网站上声明的this limitation：

带有作业类的文件被发送到 Hadoop 以运行。所以，作业文件无法尝试启动 Hadoop 作业，否则您将递归地创建 Hadoop 作业！运行作业的代码应该只能在 Hadoop 上下文之外运行。

或者，可能是因为您没有以下 (reference)：

if __name__ == '__main__':  
  MRWordCounter.run()  # where MRWordCounter is your job class

【讨论】：