0

有这个样本记录，100,1:2:3

我想标准化为
100,1
100,2
100,3

我的一位同事编写了一个猪脚本来实现这一点，而我的 MapReduce 代码花费了更多时间。我之前使用的是默认的 TextInputformat。但是为了提高性能，我决定编写一个自定义的输入格式类，带有一个自定义的 RecordReader。以 LineRecordReader 类为参考，我尝试编写以下代码。

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;

import com.normalize.util.Splitter;

public class NormalRecordReader extends RecordReader<Text, Text> {

    private long start;
    private long pos;
    private long end;
    private LineReader in;
    private int maxLineLength;
    private Text key = null;
    private Text value = null;
    private Text line = null;

    public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException {
        FileSplit split = (FileSplit) genericSplit;
        Configuration job = context.getConfiguration();
        this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength", Integer.MAX_VALUE);

        start = split.getStart();
        end = start + split.getLength();

        final Path file = split.getPath();

        FileSystem fs = file.getFileSystem(job);
        FSDataInputStream fileIn = fs.open(split.getPath());

        in = new LineReader(fileIn, job);
        this.pos = start;
    }

    public boolean nextKeyValue() throws IOException {
        int newSize = 0;
        if (line == null) {
            line = new Text();
        }

        while (pos < end) {
            newSize = in.readLine(line);
            if (newSize == 0) {
                break;
            }
            pos += newSize;
            if (newSize < maxLineLength) {
                break;
            }

            // line too long. try again
            System.out.println("Skipped line of size " + newSize + " at pos " + (pos - newSize));
        }
        Splitter splitter = new Splitter(line.toString(), ",");
        List<String> split = splitter.split();

        if (key == null) {
            key = new Text();
        }
        key.set(split.get(0));

        if (value == null) {
            value = new Text();
        }
        value.set(split.get(1));

        if (newSize == 0) {
            key = null;
            value = null;
            return false;

        } else {
            return true;
        }
    }

    @Override
    public Text getCurrentKey() {
        return key;
    }

    @Override
    public Text getCurrentValue() {
        return value;
    }

    /**
     * Get the progress within the split
     */
    public float getProgress() {
        if (start == end) {
            return 0.0f;
        } else {
            return Math.min(1.0f, (pos - start) / (float)(end - start));
        }
    }

    public synchronized void close() throws IOException {
        if (in != null) {
            in.close(); 
        }
    }
}

虽然这可行，但我没有看到任何性能改进。在这里，我打破了“，”的记录，并将 100 设置为键，将 1,2,3 设置为值。我只调用执行以下操作的映射器：

public void map(Text key, Text value, Context context) 
        throws IOException, InterruptedException {

    try {
        Splitter splitter = new Splitter(value.toString(), ":");
        List<String> splits = splitter.split();

        for (String split : splits) {
            context.write(key, new Text(split));
        }

    } catch (IndexOutOfBoundsException ibe) {
        System.err.println(value + " is malformed.");
    }
}

拆分器类用于拆分数据，因为我发现 String 的拆分器速度较慢。方法是：

public List<String> split() {

    List<String> splitData = new ArrayList<String>();
    int beginIndex = 0, endIndex = 0;

    while(true) {

        endIndex = dataToSplit.indexOf(delim, beginIndex);
        if(endIndex == -1) {
            splitData.add(dataToSplit.substring(beginIndex));
            break;
        }

        splitData.add(dataToSplit.substring(beginIndex, endIndex));
        beginIndex = endIndex + delimLength;
    }

    return splitData;
}

可以以任何方式改进代码吗？

4

1 回答 1

1

让我在这里总结一下我认为您可以改进的地方，而不是在评论中：

  • 如前所述，目前您正在为Text每条记录创建多次对象（次数将等于您的令牌数）。虽然对于少量投入来说可能并不重要，但对于体面的工作来说，这可能是一件大事。要解决此问题，请执行以下操作：

    private final Text text = new Text();

public void map(Text key, Text value, Context context) {
    ....
    for (String split : splits) {
        text.set(split);
        context.write(key, text);
    }
}

  • 对于您的拆分，您现在正在为每条记录分配一个新数组，填充该数组，然后遍历该数组以写入您的输出。实际上，在这种情况下，您实际上并不需要数组，因为您没有维护任何状态。使用split您提供的方法的实现，您只需对数据进行一次传递：

    public void map(Text key, Text value, Context context) {
    String dataToSplit = value.toString();
    String delim = ":";

    int beginIndex = 0;
    int endIndex = 0;

    while(true) {
        endIndex = dataToSplit.indexOf(delim, beginIndex);
        if(endIndex == -1) {
            text.set(dataToSplit.substring(beginIndex));
            context.write(key, text);
            break;
        }

        text.set(dataToSplit.substring(beginIndex, endIndex));
        context.write(key, text);
        beginIndex = endIndex + delim.length();
    }
}

  • 我真的不明白你为什么要编写自己的InputFormat，这似乎KeyValueTextInputFormat正是你所需要的，并且可能已经被优化了。以下是你如何使用它：

    conf.set("key.value.separator.in.input.line", ",");
job.setInputFormatClass(KeyValueTextInputFormat.class);

  • 根据您的示例，每条记录的键似乎是一个整数。如果总是这样，那么使用 aText作为您的映射器输入键并不是最佳选择，它应该是 aIntWritable甚至可能是 a ，ByteWritable具体取决于您的数据中的内容。

  • 同样，您希望使用IntWritableorByteWritable作为映射器输出键和输出值。

此外，如果您想要一些有意义的基准，您应该在更大的数据集上进行测试，如果可能的话，像几个 Gbs。1 分钟的测试并没有真正的意义，尤其是在分布式系统的上下文中。一项工作在小投入上可能比另一个工作运行得更快，但对于更大的投入，趋势可能会逆转。

话虽如此，您还应该知道 Pig 在转换为 Map/Reduce 时在后台做了很多优化，所以我并不感到惊讶它比您的 Java Map/Reduce 代码运行得更快，而且我在过去。尝试我建议的优化，如果它仍然不够快，这里有一个关于使用一些更有用的技巧分析 Map/Reduce 作业的链接（特别是关于分析的技巧 7，我发现它很有用）。

于 2013-01-20T08:03:52.120 回答