hadoop - Hadoop Streaming 作业中的并行映射器任务数

Question

我刚开始学习Hadoop。我正在尝试将流接口(interface)与处理文件的 Python 脚本结合使用:对于每个输入文件，我都会创建一个包含有关它的一些信息的输出文件，因此这是一个没有 reducer 的映射作业。我发现文件一次被处理一个，这不是我想要的。

我会解释我做了什么，但我也会在之后发布一些代码，以防我在那里遗漏了一些东西。

我有一个输入格式和记录阅读器，它可以读取整个文件并将其内容用作值，将文件名用作键。 (文件并不大。)另一方面，我有一个输出格式和记录写入器，它根据键将值写入具有名称的文件。我正在使用 -io rawbytes我的 Python 脚本知道如何读取和写入键/值对。

就产生我期望的输出而言，这一切都很好。如果我使用例如 10 个输入文件运行，我会得到 10 个拆分。这意味着每次我的脚本运行时它只会获得一个键/值对——这并不理想，但也不是什么大问题，而且我可以看到这可能是不可避免的。不太好的地方是，在任何时候都只有一个正在运行的脚本实例。设置 mapreduce.job.maps 没有任何区别(虽然我隐约记得看到关于这个值的一些东西只是一个建议，所以也许 Hadoop 正在做出不同的决定)。我错过了什么？

这是我的代码:-

#!/bin/bash

hadoop jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar \
    -libjars mimi.jar \
    -D mapreduce.job.reduces=0 \
    -files rawbytes_mapper.py,irrelevant.py \
    -inputformat "mimi.WholeFileInputFormat" \
    -outputformat "mimi.NamedFileOutputFormat" \
    -io rawbytes \
    -mapper "rawbytes_mapper.py irrelevant blah blah blah" \
    -input "input/*.xml" \
    -output output

#!/usr/bin/python

def read_raw_bytes(input):
    length_bytes = input.read(4)
    if len(length_bytes) < 4:
        return None
    length = 0
    for b in length_bytes:
        length = (length << 8) + ord(b)
    return input.read(length)

def write_raw_bytes(output, s):
    length = len(s)
    length_bytes = []
    for _ in range(4):
        length_bytes.append(chr(length & 0xff))
        length = length >> 8
    length_bytes.reverse()
    for b in length_bytes:
        output.write(b)
    output.write(s)

def read_keys_and_values(input):
    d = {}
    while True:
        key = read_raw_bytes(input)
        if key is None: break
        value = read_raw_bytes(input)
        d[key] = value
    return d

def write_keys_and_values(output, d):
    for key in d:
        write_raw_bytes(output, key)
        write_raw_bytes(output, d[key])

if __name__ == "__main__":
    import sys
    module = __import__(sys.argv[1])
    before = read_keys_and_values(sys.stdin)
    module.init(sys.argv[2:])
    after = module.process(before)
    write_keys_and_values(sys.stdout, after)

package mimi;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;

public class WholeFileInputFormat extends FileInputFormat<BytesWritable, BytesWritable>
{
    private static class WholeFileRecordReader implements RecordReader<BytesWritable, BytesWritable>
    {
        private FileSplit split;
        private JobConf conf;
        private boolean processed = false;

        public WholeFileRecordReader(FileSplit split, JobConf conf)
        {
            this.split = split;
            this.conf = conf;
        }

        @Override
        public BytesWritable createKey()
        {
            return new BytesWritable();
        }

        @Override
        public BytesWritable createValue()
        {
            return new BytesWritable();
        }

        @Override
        public boolean next(BytesWritable key, BytesWritable value) throws IOException
        {
            if (processed)
            {
                return false;
            }

            byte[] contents = new byte[(int) split.getLength()];
            Path file = split.getPath();
            String name = file.getName();
            byte[] bytes = name.getBytes(StandardCharsets.UTF_8);
            key.set(bytes, 0, bytes.length);
            FileSystem fs = file.getFileSystem(conf);
            FSDataInputStream in = null;
            try
            {
                in = fs.open(file);
                IOUtils.readFully(in, contents, 0, contents.length);
                value.set(contents, 0, contents.length);
            }
            finally
            {
                IOUtils.closeStream(in);
            }

            processed = true;
            return true;
        }

        @Override
        public float getProgress() throws IOException
        {
            return processed ? 1.0f : 0.0f;
        }

        @Override
        public long getPos() throws IOException
        {
            return processed ? 0l : split.getLength();
        }

        @Override
        public void close() throws IOException
        {
            // do nothing
        }
    }

    @Override
    protected boolean isSplitable(FileSystem fs, Path file)
    {
        return false;
    }

    @Override
    public RecordReader<BytesWritable, BytesWritable> getRecordReader(InputSplit split,
                                                                      JobConf conf,
                                                                      Reporter reporter)
    throws IOException
    {
        return new WholeFileRecordReader((FileSplit) split, conf);
    }
}

package mimi;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.lib.MultipleOutputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.Progressable;

public class NamedFileOutputFormat extends MultipleOutputFormat<BytesWritable, BytesWritable>
{
    private static class BytesValueWriter implements RecordWriter<BytesWritable, BytesWritable>
    {
        FSDataOutputStream out;

        BytesValueWriter(FSDataOutputStream out)
        {
            this.out = out;
        }

        @Override
        public synchronized void write(BytesWritable key, BytesWritable value) throws IOException
        {
            out.write(value.getBytes(), 0, value.getLength());
        }

        @Override
        public void close(Reporter reporter) throws IOException
        {
            out.close();
        }
    }

    @Override
    protected String generateFileNameForKeyValue(BytesWritable key, BytesWritable value, String name)
    {
        return new String(key.getBytes(), 0, key.getLength(), StandardCharsets.UTF_8);
    }

    @Override
    public RecordWriter<BytesWritable, BytesWritable> getBaseRecordWriter(FileSystem ignored,
                                                                          JobConf conf,
                                                                          String name,
                                                                          Progressable progress)
    throws IOException
    {
        Path file = FileOutputFormat.getTaskOutputPath(conf, name);
        FileSystem fs = file.getFileSystem(conf);
        FSDataOutputStream out = fs.create(file, progress);
        return new BytesValueWriter(out);
    }
}

Answer 1

我想我可以帮助你解决这部分问题:

each time my script runs it only gets one key/value pair - which isn't ideal

如果 isSplitable 方法返回 false，则每个映射器只会处理一个文件。因此，如果您不覆盖 isSplitable 方法并让它返回 true，您应该在一个映射器中拥有多个键/值对。在您的情况下，每个文件都是一个键/值对，因此即使 isSplitable 返回 true 也无法拆分它们。

我无法弄清楚为什么一次只启动一个映射器，但我仍在考虑它:)