使用MapReduce实现java规范化

Question

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

我想将其标准化为，
100,1
100,2
100,3

我的一位同事编写了一个pig脚本来实现这一点，我的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.");
    }
}

splitter类用于分割数据，因为我发现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;
}

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

Answer 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);
  

• 根据您的示例，每条记录的键似乎是一个整数。如果总是这样，那么使用Text作为映射器输入键不是最佳的，它应该是IntWritable或者甚至是ByteWritable，这取决于数据中的内容

• 类似地，您希望使用IntWritable或ByteWritable作为映射器输出键和输出值

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

话虽如此，您也应该知道Pig在转换为Map/Reduce时做了大量的幕后优化，所以我对它比Java Map/Reduce代码运行得更快并不感到惊讶，我在过去也看到过这一点。尝试我建议的优化，如果它仍然不够快，这里是a link on profiling your Map/Reduce jobs，还有一些更有用的技巧（特别是关于分析的技巧7，我发现这很有用）