通过Python在Hadoop流中使用文件

Question

我对Hadoop和MapReduce完全陌生，正在努力完成它。我正在尝试用python开发一个mapduce应用程序，在这个应用程序中，我使用来自2. CSV文件的数据。我只是在mapper中读取这两个文件，然后将文件中的键值对打印到sys.stdout

当我在一台机器上使用它时，该程序运行正常，但是使用Hadoop流式处理时，我遇到了错误。我认为我在Hadoop上的映射器中读取文件时犯了一些错误。请帮助我编写代码，并告诉我如何在Hadoop流中使用文件处理。mapper.py 代码如下。（您可以从注释中理解代码）：

#!/usr/bin/env python
import sys
from numpy import genfromtxt

def read_input(inVal):
    for line in inVal:
        # split the line into words
        yield line.strip()

def main(separator='\t'):
    # input comes from STDIN (standard input)
    labels=[]
    data=[]    
    incoming = read_input(sys.stdin)
    for vals in incoming:
        # write the results to STDOUT (standard output);
        # what we output here will be the input for the
        # Reduce step, i.e. the input for reducer.py
        #
        # tab-delimited;
        if len(vals) > 10:
            data.append(vals)
        else:
            labels.append(vals)

    for i in range(0,len(labels)):
        print "%s%s%s\n" % (labels[i], separator, data[i])


if __name__ == "__main__":
    main()

有 60000 条记录从两个.csv文件输入到此映射器，如下所示（在单台计算机上，而不是在 hadoop 群集上）：

cat mnist_train_labels.csv mnist_train_data.csv | ./mapper.py

Answer 1

你没有发布你的错误。在流式传输中，您需要传递-file参数或-input，以便文件可以随流式传输作业一起上传，或者知道在hdfs上的何处可以找到它。

Answer 2

在寻找解决方案3天后，我终于解决了这个问题。

问题出在新版本的Hadoop上（在我的例子中是2.2.0）。当从文件中读取值时，映射器代码在某个时候给出了非零的退出代码（可能是因为它一次读取了一个巨大的值列表（784））。Hadoop 2.2.0中有一个设置，它告诉Hadoop系统给出一个一般错误（代码1的子进程失败）。默认情况下，此设置设置为True。我只需要将此属性的值设置为False，它使我的代码运行没有任何错误。

设置为：流非零、退出失败。只需在流式传输时将其设置为 false 即可。因此，流式处理命令将有点像：

**hadoop jar ... -D stream.non.zero.exit.is.failure=false ...**

希望它能帮助某人，并节省3天... ;)