3

私はHadoop0.20.2を使用しており、古いAPIを使用しています。一度に1行を送信するのではなく、データのチャンクをマッパーに送信しようとしています（データは複数の行をカバーしています）。NLineInputFormatを使用して、一度に取得する行数を設定しようとしましたが、マッパーは一度に1行しか受信していません。私は正しいコードを持っていると確信しています。これが機能しない理由はありますか？

ご参考までに、

JobConf conf = new JobConf（WordCount.class）;

conf.setInt（ "mapred.line.input.format.linespermap"、2）;

conf.setInputFormat（NLineInputFormat.class）;

基本的に、私はhttp://hadoop.apache.org/common/docs/r0.20.2/mapred_tutorial.html#Example%3A+WordCount+v1.0のサンプルコードを使用しており、TextInputFormatのみを変更しています。

前もって感謝します

4

2 に答える 2

4

NLineInputFormatは、マッパーがすべて同じ数の入力レコードを受け取るように設計されています（各ファイルの分割の最後の部分を除く）。

したがって、入力プロパティを2に変更することにより、各マッパーは（最大で）一度に2つの入力行ではなく、2つの入力ペアを受け取る必要があります（これは私が探しているものです）。

これは、各マップタスクのカウンター「マップ入力レコード」を確認することで確認できるはずです。これは、ほとんどのマッパーについて2を報告するはずです。

于 2012-06-12T18:01:09.927 に答える
0

最近、NLineInputFormatをオーバーライドし、デフォルトのLineReaderの代わりにカスタムMultiLineRecordReaderを実装する独自のInputFormatを作成することで、この問題を解決しました。

分割ごとに正確にN行になるという同じ保証が必要だったため、NLineInputFormatを拡張することを選択しました。

このレコードリーダーは、http：//bigdatacircus.com/2012/08/01/wordcount-with-custom-record-reader-of-textinputformat/からほぼそのまま取得されます。

私が変更したのは、そのプロパティがmaxLineLength新しいAPIを使用することと、その値がハードコーディングされているNLINESTOPROCESSNLineInputFormatの内部から読み取られることsetNumLinesPerSplit()だけです（柔軟性を高めるため）。

結果は次のとおりです。

public class MultiLineInputFormat extends NLineInputFormat{
    @Override
    public RecordReader<LongWritable, Text> createRecordReader(InputSplit genericSplit, TaskAttemptContext context) {
        context.setStatus(genericSplit.toString());
        return new MultiLineRecordReader();
    }

    public static class MultiLineRecordReader extends RecordReader<LongWritable, Text>{
        private int NLINESTOPROCESS;
        private LineReader in;
        private LongWritable key;
        private Text value = new Text();
        private long start =0;
        private long end =0;
        private long pos =0;
        private int maxLineLength;

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

        @Override
        public LongWritable getCurrentKey() throws IOException,InterruptedException {
            return key;
        }

        @Override
        public Text getCurrentValue() throws IOException, InterruptedException {
            return value;
        }

        @Override
        public float getProgress() throws IOException, InterruptedException {
            if (start == end) {
                return 0.0f;
            }
            else {
                return Math.min(1.0f, (pos - start) / (float)(end - start));
            }
        }

        @Override
        public void initialize(InputSplit genericSplit, TaskAttemptContext context)throws IOException, InterruptedException {
            NLINESTOPROCESS = getNumLinesPerSplit(context);
            FileSplit split = (FileSplit) genericSplit;
            final Path file = split.getPath();
            Configuration conf = context.getConfiguration();
            this.maxLineLength = conf.getInt("mapreduce.input.linerecordreader.line.maxlength",Integer.MAX_VALUE);
            FileSystem fs = file.getFileSystem(conf);
            start = split.getStart();
            end= start + split.getLength();
            boolean skipFirstLine = false;
            FSDataInputStream filein = fs.open(split.getPath());

            if (start != 0){
                skipFirstLine = true;
                --start;
                filein.seek(start);
            }
            in = new LineReader(filein,conf);
            if(skipFirstLine){
                start += in.readLine(new Text(),0,(int)Math.min((long)Integer.MAX_VALUE, end - start));
            }
            this.pos = start;
        }

        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            if (key == null) {
                key = new LongWritable();
            }
            key.set(pos);
            if (value == null) {
                value = new Text();
            }
            value.clear();
            final Text endline = new Text("\n");
            int newSize = 0;
            for(int i=0;i<NLINESTOPROCESS;i++){
                Text v = new Text();
                while (pos < end) {
                    newSize = in.readLine(v, maxLineLength,Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),maxLineLength));
                    value.append(v.getBytes(),0, v.getLength());
                    value.append(endline.getBytes(),0, endline.getLength());
                    if (newSize == 0) {
                        break;
                    }
                    pos += newSize;
                    if (newSize < maxLineLength) {
                        break;
                    }
                }
            }
            if (newSize == 0) {
                key = null;
                value = null;
                return false;
            } else {
                return true;
            }
        }
    }

}
于 2015-04-29T06:27:55.727 に答える