MapReduce求各年销售笔数、销售总额实验（流程+代码）

流程：

1. 读取销售数据文件，将数据按照年份划分为多个小数据集。
2. 对每个小数据集进行Map操作，将每个订单的销售笔数和销售额作为键值对输出。
3. 对每个小数据集的Map输出进行Reduce操作，将相同键的值进行累加，得到每年的销售笔数和销售总额。
4. 输出每年的销售笔数和销售总额。

代码：

Map阶段：

public class SalesMap extends Mapper<Object, Text, IntWritable, DoubleWritable> {
    private final static IntWritable year = new IntWritable();
    private final static DoubleWritable sales = new DoubleWritable();

    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        String[] fields = value.toString().split(",");
        int orderYear = Integer.parseInt(fields[0].substring(0, 4));
        double orderSales = Double.parseDouble(fields[1]);
        year.set(orderYear);
        sales.set(orderSales);
        context.write(year, sales);
    }
}

Reduce阶段：

public class SalesReduce extends Reducer<IntWritable, DoubleWritable, IntWritable, Text> {
    private final static DecimalFormat decimalFormat = new DecimalFormat("#.00");

    public void reduce(IntWritable key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
        int orderCount = 0;
        double orderSales = 0.0;
        for (DoubleWritable value : values) {
            orderCount++;
            orderSales += value.get();
        }
        String salesStr = decimalFormat.format(orderSales);
        Text salesText = new Text("订单数：" + orderCount + " 销售总额：" + salesStr);
        context.write(key, salesText);
    }
}

Driver阶段：

public class SalesDriver extends Configured implements Tool {
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(), "Sales");
        job.setJarByClass(SalesDriver.class);
        job.setMapperClass(SalesMap.class);
        job.setReducerClass(SalesReduce.class);
        job.setOutputKeyClass(IntWritable.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;
    }

    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new SalesDriver(), args);
        System.exit(exitCode);
    }
}