1

我有一个 map reduce 程序正在运行以读取 HDFS 文件，如下所示：

hadoop jar /opt/mapr/hadoop/hadoop-0.20.2/contrib/streaming/hadoop-0.20.2-dev-streaming.jar -Dmapred.reduce.tasks=1000  -file $homedir/mapper.py -mapper $homedir/mapper.py -file $homedir/reducer.py -reducer $homedir/reducer.py   -input /user/data/* -output /output/ 2> output.text

有什么需要确认的，路径 /user/data/* 有文件夹，包括文件， /user/data/* 将迭代所有子文件夹下的所有文件，对吗？

hdfs 文本文件的每一行都包含一个 JSON 字符串，因此映射器读取文件如下：

for line in sys.stdin:
    try:
        object = json.loads(line)

但是 HDFS 的所有者将文件从文本更改为序列文件。我发现 map reduce 程序输出了很多零大小的文件，这可能意味着它没有成功地从 HDFS 读取文件。

我应该对代码进行哪些更改以便可以从序列文件中读取？我还有一个 HIVE 外部表来根据 mapreduce 的输出执行聚合和排序，并且 HIVE 之前是 STORED AS TEXTFILE ，我应该更改为 STORED AS SEQUENCEFILE 吗？

谢谢，

4

1 回答 1

4

看看这个

在您的 mapreduce 作业输入之前运行以下 python 文件
：您的序列文件
输出：您对 mapreduce 的输入

import sys

from hadoop.io import SequenceFile

if __name__ == '__main__':
    if len(sys.argv) < 3:
        print 'usage: SequenceFileReader <filename> <output>'
    else:
        reader = SequenceFile.Reader(sys.argv[1])

    key_class = reader.getKeyClass()
    value_class = reader.getValueClass()

    key = key_class()
    value = value_class()

    #reader.sync(4042)
    position = reader.getPosition()
    f = open(sys.argv[2],'w')
    while reader.next(key, value):
        f.write(value.toString()+'\n')
    reader.close()
    f.close()

您现在不必更改原始 python 文件。

于 2015-11-13T01:58:17.617 回答