查找我们所有商店中按产品类别划分的销售明细

Question

我有一个销售文件，其中包含商店名称、位置、销售价格、产品名称等信息。文件格式如下所示，2012-01-01  09:00   San Jose    Men's Clothing  214.05  Amex2012-01-01  09:00   Fort Worth  Women's Clothing    153.57  Visa2012-01-01  09:00   San Diego   Music   66.08   Cash2012-01-01  09:00   Pittsburgh  Pet Supplies    493.51  Discover2012-01-01  09:00   Omaha   Children's Clothing 235.63  MasterCard2012-01-01  09:00   Stockton    Men's Clothing  247.18  MasterCard  我想编写一个 Map-reduce 作业来查找我们所有商店中按产品类别划分的销售明细。下面提供了我的代码（包括 Mapper 和 reducer），public final class P1Q1 {    public static final class P1Q1Map extends Mapper<LongWritable, Text, Text, DoubleWritable> {        private final Text word = new Text();        public final void map(final LongWritable key, final Text value, final Context context)                throws IOException, InterruptedException {            final String line = value.toString();            final String[] data = line.trim().split("\t");            if (data.length == 6) {                final String product = data[3];                final double sales = Double.parseDouble(data[4]);                word.set(product);                context.write(word, new DoubleWritable(sales));            }        }    }    public static final class P1Q1Reduce extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {        public final void reduce(final Text key, final Iterable<DoubleWritable> values, final Context context)                throws IOException, InterruptedException {            double sum = 0.0;            for (final DoubleWritable val : values) {                sum += val.get();            }            context.write(key, new DoubleWritable(sum));        }    }}代码提供的答案不正确，与 Udacity 结果不匹配。任何人都知道这是否是正确的想法以及如何做到这一点？

Answer 1

在大多数情况下，我会说您的代码看起来不错，并且组合器只是一种优化，因此排除它应该产生与包含它相同的输出。

我写了我自己的 MR，我得到了给定输入的输出

Children's Clothing 235.63

Men's Clothing  461.23

Music   66.08

Pet Supplies    493.51

Women's Clothing    153.57

显然，如果您有成百上千的商店，那么您将获得数百万个货币单位，如您的输出所示。

代码

@Override

public int run(String[] args) throws Exception {

    Configuration conf = getConf();

    Job job = Job.getInstance(conf, APP_NAME);

    job.setJarByClass(StoreSumRunner.class);

    job.setMapperClass(TokenizerMapper.class);

    job.setReducerClass(CurrencyReducer.class);

    job.setOutputKeyClass(Text.class);

    job.setOutputValueClass(DoubleWritable.class);

    FileInputFormat.addInputPath(job, new Path(args[0]));

    FileOutputFormat.setOutputPath(job, new Path(args[1]));

    return job.waitForCompletion(true) ? 0 : 1;

}

static class TokenizerMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> {

    private final Text key = new Text();

    private final DoubleWritable sales = new DoubleWritable();

    @Override

    protected void map(LongWritable offset, Text value, Context context) throws IOException, InterruptedException {

        final String line = value.toString();

        final String[] data = line.trim().split("\\s\\s+");

        if (data.length < 6) {

            System.err.printf("mapper: not enough records for %s%n", Arrays.toString(data));

            return;

        }

        key.set(data[3]);

        try {

            sales.set(Double.parseDouble(data[4]));

            context.write(key, sales);

        } catch (NumberFormatException ex) {

            System.err.printf("mapper: invalid value format %s%n", data[4]);

        }

    }

}

static class CurrencyReducer extends Reducer<Text, DoubleWritable, Text, Text> {

    private final Text output = new Text();

    private final DecimalFormat df = new DecimalFormat("#.00");

    @Override

    protected void reduce(Text date, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {

        double sum = 0;

        for (DoubleWritable value : values) {

            sum += value.get();

        }

        output.set(df.format(sum));

        context.write(date, output);

    }

}