java - 映射器的意外输出。它在输出前添加一个数字

Question

因此，我将来自另一个 MapReduce 作业的输入提供给我的 Mapper。在这里，我对我的输入进行了一些分区，以便 reducer 可迭代对象不会超出内存(这只是一个测试程序)。所以在映射器中，我只是试图删除输入中的“/”，然后在缩减器中添加总和，但是映射器开始给出一个不寻常的输出，它在输出前添加了一个整数，而其余输出是也不尽如人意。 同样在此之前，我收到了类似预期的错误 'org.apache.hadoop.io.Text, received org.apache.hadoop.io.LongWritable' 并添加了这个 ' job1.setMapOutputKeyClass( LongWritable.class);
job1.setMapOutputValueClass(Text.class);' 避免了错误。

如果我在某处严重错误，请原谅。

package test;
import java.io.IOException;
import java.util.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
public class MapCom2
{
    public static class Map1 extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable();
    private Text word = new Text();
    public static int cnt=1;
    public void map1(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        {
        String line = value.toString();
        Configuration conf=context.getConfiguration();
        String []tokens=line.split("\t");
        int l=0;
        while(l<tokens.length)
            {
            if(tokens[l].contains("/"))
                break;
            l=l+1;
            }
        int indno=tokens[l].lastIndexOf("/");
        String w=tokens[l].substring(0,indno);
        int tcnt=Integer.parseInt(tokens[l+1]);
        word.set(w);
        one.set(tcnt);
        context.write(word,one);
        }
    }

    public static int main(String[] args) throws Exception {
    Configuration conf1= new Configuration();
    Job job1 = new Job(conf1,"mapcom2");
    job1.setJarByClass(test.MapCom2.class);
    job1.setJobName("mapcom2");
    job1.setMapOutputKeyClass(LongWritable.class);
    job1.setMapOutputValueClass(Text.class);
    job1.setOutputKeyClass(Text.class);
    job1.setOutputValueClass(IntWritable.class);
    job1.setMapperClass(Map1.class);
    //job1.setReducerClass(Reduce1.class);
    String op=args[0];
    if(!(op.charAt(op.length()-1)+"").equals("/"))
        op=op+"/"+"part-r-00000";
    else
        op=op+"part-r-00000";
    job1.setInputFormatClass(TextInputFormat.class);
    job1.setOutputFormatClass(TextOutputFormat.class);
    FileInputFormat.addInputPath(job1, new Path(op));
    FileOutputFormat.setOutputPath(job1, new Path(args[1]));
    int ret=job1.waitForCompletion(true)?0:1;
    return ret;
    }
}

输入是这样的。

)hadoop/0   1
-C/0    1
-classpath/0    1
-cvf/0  1
-d/0    1
-mkdir/0    2
-put/0  2
./0 1
/home/hadoop/hadoop/0   1
/home/hadoop/hadoopAssign/wordcount_classes/0   1
/wordcount/input/0  3
/wordcount/output/0 1
1)/0    1
2)/0    1
3)/0    1
4)/0    1
5/0 1
Assign/hadoop-core-1.1.2.jar/0  1
WordCount.java/0    1
and/0   1
copy/0  1
file01/0    2
file02/0    2
files/0 1
fs/0    3
fs/1    2
hadoop/0    3
hadoop/1    2
jar/0   2
javac/0 1
make/0  1
mkdir/0 1
org.myorg.WordCount/0   1
them/0  1
to/0    2
two/0   1
wordcount.jar/0 2
wordcount/0 1
wordcount/input/0   1
wordcount_classes//0    1
wordcount_classes/0 1

而输出是这样的

0   )hadoop/0   1
12  -C/0    1
19  -classpath/0    1
34  -cvf/0  1
43  -d/0    1
50  -mkdir/0    2
61  -put/0  2
70  ./0 1
76  /home/hadoop/hadoop/0   1
100 /home/hadoop/hadoopAssign/wordcount_classes/0   1
148 /wordcount/input/0  3
169 /wordcount/output/0 1
191 1)/0    1
198 2)/0    1
205 3)/0    1
212 4)/0    1
219 5/0 1
225 Assign/hadoop-core-1.1.2.jar/0  1
258 WordCount.java/0    1
277 and/0   1
285 copy/0  1
294 file01/0    2
305 file02/0    2
316 files/0 1
326 fs/0    3
333 fs/1    2
340 hadoop/0    3
351 hadoop/1    2
362 jar/0   2
370 javac/0 1
380 make/0  1
389 mkdir/0 1
399 org.myorg.WordCount/0   1
423 them/0  1
432 to/0    2
439 two/0   1
447 wordcount.jar/0 2
465 wordcount/0 1
479 wordcount/input/0   1
499 wordcount_classes//0    1
522 wordcount_classes/0 1

前几行的预期输出是这样的

)hadoop 1
-C  1
-classpath  1
-cvf    1

这正是我想要做的，但问题出在上面的程序中。我的最终目标是限制 Reducer 中可迭代值的大小。对于单词 hadoop 和 fs，我们将输出 5 和 5。在这里，我通过以某种方式对映射器进行分区将 reducer 值限制为 3，但我卡在了我的映射器中。所以/0、/1、/2 只是表示例如单词“hadoop”已经出现了 3 次，并且所以我们转到 hadoop/1 。这是在前面的 map reduce 程序中完成的，稍后我将介绍线性链。感谢任何帮助。

Answer 1

在 Hadoop 框架中，映射阶段被分配给一个映射器作业。这一个与您需要实现的特定接口(interface)相关联:Mapper。

此接口(interface)有一个名为 map 的独特方法，您需要实现该方法才能正确实现映射阶段。

您的代码中有错字 (map1)。更正它，用户已经解决了它的问题。