如何在hadoop中将映射器键作为文本输出到Reducer

Question

我正在尝试将 String 转换为 Text 对象并将值作为 MapOutputKey 输出到 Reducer。但是，在 reducer 中，键看起来像一个黑色空间，其中 value（不是 MapOutputValue）作为内存位置。 但是，如果我将字符串硬编码到文本对象中并将其发送出去，我就可以在 Reducer 中看到正确的值。 编辑：-现在这很奇怪也很有趣。相同的程序适用于较小的数据集，即从映射器发送的文本键在减速器中接收。但是如果我使用原始数据集，从 map 发送的文本键不会反映在 reducer 中。 我不知道，关于 Hadoop 有一些我不知道的东西。 这是完整的源代码：-

 public class InvertedIndex {
    public InvertedIndex(){}
    public static class InvertedMap extends Mapper<IntWritable,Text,Text,DocAttributes> {
        Integer docNum;
        String word;
        Integer docFrequency;
        Integer termFrequency;
        private Text key = new Text();
        @Override
        public void map(IntWritable mapIn,Text mapValIn,Context context) throws IOException,InterruptedException{
            try{
                String line = mapValIn.toString();
                String[] words = line.trim().split("\\s");
                List<String> wordList = new ArrayList<String>(Arrays.asList(words));
                String k;
                for(int i=0; i<wordList.size();i++){
                   k = wordList.get(i);
                    //Text key = new Text(k);
                  //  DocAttributes da = new DocAttributes();
                    key.set(k);
                    int sum=1;
                    for(int j=i;j<wordList.size();j++){

                        if(wordList.get(i).matches(wordList.get(j)) && j>i){
                            sum++;
                            docNum = mapIn.get();
                            docFrequency = sum;
                            word = wordList.get(i);
                            termFrequency = sum;
                            wordList.remove(j);
                        }


                        else{
                            docNum = mapIn.get();
                            docFrequency = sum;
                            word = wordList.get(i);
                            termFrequency = sum;
                        }

                    }
                    if(i == words.length-1){
                        docNum = mapIn.get();
                        docFrequency = sum;
                        word = wordList.get(i);
                        termFrequency = sum;
                    }
                    context.write(key, new DocAttributes(docNum,word,docFrequency,termFrequency));
                }
            }
            catch(NullPointerException ne){
                ne.printStackTrace();
            }
        }

    }
    public static class InvertedReduce extends Reducer<Text,DocAttributes,LongWritable,DocAttributes>{
       @Override
        public void reduce(Text key, Iterable<DocAttributes> value,Context context) throws IOException,InterruptedException{
            Iterator<DocAttributes> iterator = value.iterator();
            DocAttributes doc = new DocAttributes();
            List<DocAttributes> list = new ArrayList<DocAttributes>();
            while(iterator.hasNext()){
                list.add(new DocAttributes(iterator.next()));
            }
            Integer docFrequency = 0;
            for(DocAttributes d : list){
                docFrequency += d.getDocFrequency();
                doc.setDocNum(d.getDocNum());
                doc.setWord(d.getWord());
                doc.setDocFrequency(docFrequency);
                doc.setTermFrequency(d.getTermFrequency());
           }

           context.write(new LongWritable(), doc);

       }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setMapperClass(InvertedMap.class);
        // job.setCombinerClass(InvertedCombine.class);
        job.setReducerClass(InvertedReduce.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DocAttributes.class);

        job.setJarByClass(InvertedIndex.class);

        TextInputFormat.addInputPath(job, new Path(args[0]));
        TextOutputFormat.setOutputPath(job,new Path(args[1]));

        job.setInputFormatClass(DocInput.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        job.submit();
        job.waitForCompletion(true);


    }

}

Answer 1

您正在使用 for 每个循环在迭代时修改 ArrayList 'wordList'。您可以尝试使用迭代器循环在迭代时删除元素 (Iterator.remove())。