Hadoop reducer 字符串操作不起作用

Question

嗨Reduce 阶段的文本操作似乎无法正常工作。我怀疑问题可能出在我的代码中而不是 hadoop 本身，但你永远不知道......如果你能发现任何陷阱，请告诉我。我浪费了一天的时间试图弄清楚这段代码有什么问题。

我的示例输入文件名为 simple.psv

12345   abc@bbc.com|m|1975
12346   bbc@cde.com|m|1981

我的 Mapper 和 reducer 代码

package simplemapreduce;

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.lib.KeyFieldBasedComparator;

/**
 *
 * @author
 */
public class Main {


    public static class SimpleMap extends MapReduceBase implements
            Mapper<LongWritable, Text, Text, Text> {

        public void map(LongWritable key, Text inputs,
                OutputCollector<Text, Text> output, Reporter reporter)
                throws IOException {

            String inputString = inputs.toString();
            //System.out.println("CRM Map record:"+inputString);
            StringTokenizer tokenizer = new StringTokenizer(inputString);
            Text userid = new Text();
            if (tokenizer.hasMoreTokens()) {
                userid.set(tokenizer.nextToken());
                Text data = new Text();
                if (tokenizer.hasMoreTokens()) {
                    data.set(tokenizer.nextToken());
                } else {
                    data.set("");
                }
                output.collect(userid, data);
            }
        }
    }

    /**
     * A reducer class that just emits its input.
     */
    public static class SimpleReduce extends MapReduceBase implements
            Reducer<Text, Text, Text, Text> {

        public void reduce(Text key, Iterator<Text> values,
                OutputCollector<Text, Text> output, Reporter reporter)
                throws IOException {

            while (values.hasNext()) {
                Text txt = values.next();
                String inputString = "<start>" + txt.toString() + "<end>";
                Text out = new Text();
                out.set(inputString);
                //System.out.println(inputString);
                output.collect(key, out);

            }
        }
    }

    public static void main(String[] args) throws IOException {

        if (args.length != 2) {
            System.err.println("Usage: SimpleMapReduce <input path> <output path>");
            System.exit(1);
        }
        JobConf conf = new JobConf(Main.class);
        conf.setJobName("Simple Map reducer");

        FileInputFormat.setInputPaths(conf, new Path(args[0]));
        FileOutputFormat.setOutputPath(conf, new Path(args[1]));
        conf.setMapperClass(SimpleMap.class);
        conf.setCombinerClass(SimpleReduce.class);
        conf.setReducerClass(SimpleReduce.class);
        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(Text.class);
        conf.setNumReduceTasks(1);
        JobClient.runJob(conf);
    }
}

我的示例启动脚本名为 simple.sh

#!/bin/bash

hadoop jar SimpleMapReduce.jar \
  /full/path/to/input/simple.tsv  /user/joebloggs/output

预期输出

12345   <start>abc@bbc.com|m|1975<end>
12346   <start>bbc@cde.com|m|1981<end>

实际输出

12345   <start><start>abc@bbc.com|m|1975<end><end>
12346   <start><start>bbc@cde.com|m|1981<end><end>

我在 Amazon s3 和 Linux 上测试了这个如果您能发现问题并让我知道是什么...那真的可以让我头皮发麻!

Answer 1

通过系统的基本数据流是:

Input -> Map -> Reduce -> output.

作为性能优化，添加了组合器以允许计算机(hadoop 集群中的许多计算机之一)在将数据传输到运行实际 reducer 的系统之前对数据进行部分聚合。

在字数统计示例中，可以从这些值开始:

1 1 1 1 1 1 1 1 1 1

将它们组合成

3 4 2 1

将它们归约成最终结果

10

所以combiner本质上是一种性能优化。如果你没有指定一个组合器，它不会改变通过的信息(即它是一个“身份缩减器”)。因此，如果数据集以这种方式保持有效，则只能将 SAME 类同时用作合并器和缩减器。在您的情况下:那不是真的 --> 您的数据现在无效。

你这样做:

conf.setCombinerClass(SimpleReduce.class);
conf.setReducerClass(SimpleReduce.class);

所以这会使您的映射器的输出通过您的 reducer 两次。第一个添加:“开始”和“结束”第二个再次添加“开始”和“结束”。

简单的解决方案:

// conf.setCombinerClass(SimpleReduce.class);
conf.setReducerClass(SimpleReduce.class);

HTH