用Java编写MapReduce对2月 商品A 325 1月 商品A 235 1月 商品B 234 2月 商品C 572 3月 商品A 123 1月 商品C 321 2月 商品B 573 3月 商品C 876进行排序，要求先按月份升序排序，再按销售量降序排序，结果输出的是：1月 商品C 321 1月 商品A 235 1月 商品B 234 2月 商品B 573 2月 商品C 572 2月 商品A 325 3月 商品C 876 3月 商品B 648 3月 商品A 123

首先，需要定义一个自定义的Writable类型用于存储每个输入记录的月份、商品名称和销售量。代码如下：

public class SalesRecord implements WritableComparable<SalesRecord> {

    private Text month;
    private Text product;
    private IntWritable sales;

    public SalesRecord() {
        month = new Text();
        product = new Text();
        sales = new IntWritable();
    }

    public SalesRecord(String month, String product, int sales) {
        this.month = new Text(month);
        this.product = new Text(product);
        this.sales = new IntWritable(sales);
    }

    public void write(DataOutput out) throws IOException {
        month.write(out);
        product.write(out);
        sales.write(out);
    }

    public void readFields(DataInput in) throws IOException {
        month.readFields(in);
        product.readFields(in);
        sales.readFields(in);
    }

    public int compareTo(SalesRecord other) {
        int cmp = month.compareTo(other.month);
        if (cmp != 0) {
            return cmp;
        }
        return -sales.compareTo(other.sales);
    }

    public Text getMonth() {
        return month;
    }

    public Text getProduct() {
        return product;
    }

    public IntWritable getSales() {
        return sales;
    }

    public void set(String month, String product, int sales) {
        this.month = new Text(month);
        this.product = new Text(product);
        this.sales = new IntWritable(sales);
    }

    public String toString() {
        return month + " " + product + " " + sales;
    }
}

然后，需要编写Mapper和Reducer类。Mapper类将每个输入记录解析为一个SalesRecord对象，并将其输出。Reducer类将所有具有相同月份的SalesRecords收集在一起，并对它们按销售量进行排序。代码如下：

public class SalesSorter {

    public static class SalesMapper extends Mapper<Object, Text, SalesRecord, NullWritable> {

        private SalesRecord salesRecord = new SalesRecord();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] fields = value.toString().split(" ");
            salesRecord.set(fields[1], fields[2], Integer.parseInt(fields[3]));
            context.write(salesRecord, NullWritable.get());
        }
    }

    public static class SalesReducer extends Reducer<SalesRecord, NullWritable, SalesRecord, NullWritable> {

        public void reduce(SalesRecord key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            for (NullWritable value : values) {
                context.write(key, NullWritable.get());
            }
        }
    }

}

最后，需要在main函数中配置MapReduce作业，并运行它。代码如下：

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

    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "sales sort");
    job.setJarByClass(SalesSorter.class);
    job.setMapperClass(SalesMapper.class);
    job.setReducerClass(SalesReducer.class);
    job.setMapOutputKeyClass(SalesRecord.class);
    job.setMapOutputValueClass(NullWritable.class);
    job.setOutputKeyClass(SalesRecord.class);
    job.setOutputValueClass(NullWritable.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
}

将以上所有代码保存为一个Java文件并编译运行，就可以得到按月份升序排序、按销售量降序排序的结果。