使用 Python 在 Hadoop Streaming 中使用文件

Question

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

当我在单台机器上使用该程序时，它运行良好，但使用 Hadoop Streaming，我得到一个错误。我认为我在 Hadoop 上的映射器中读取文件时犯了一些错误。请帮我写代码，并告诉我如何在 Hadoop Streaming 中使用文件处理。 mapper.py 代码如下。 （从cmets可以看懂代码）：

#!/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，它使我的代码运行时没有任何错误。

设置为：stream.non.zero.exit.is.failure。流式传输时只需将其设置为 false 。所以流式传输命令有点像：

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

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