已经知道商品和销量 如何用javamapreduce统计销量前十

可以采用MapReduce来实现统计销量前十的功能。

首先，对于每个商品，我们需要将其销量进行累加。因此，Map阶段的输入应该是（商品名称，销量）键值对。在Map阶段，我们可以将每个商品名称作为键，将其对应的销量作为值，然后输出（商品名称，销量）键值对。

接着，我们需要对所有商品的销量进行汇总。Reduce阶段的输入应该是（商品名称，销量列表）键值对。在Reduce阶段，我们可以将所有销量进行求和，然后将结果输出为（商品名称，销量总和）键值对。

最后，我们需要对所有商品按照销量进行排序，取出销量前十的商品即可。这个过程可以在客户端进行。

下面是示例代码：

Map阶段：

public static class SalesMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

  private Text outKey = new Text();
  private IntWritable outValue = new IntWritable();

  public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String line = value.toString();
    String[] fields = line.split(",");

    String productName = fields[0];
    int sales = Integer.parseInt(fields[1]);

    outKey.set(productName);
    outValue.set(sales);

    context.write(outKey, outValue);
  }
}

Reduce阶段：

public static class SalesReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

  public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    int sum = 0;
    for (IntWritable value : values) {
      sum += value.get();
    }
    context.write(key, new IntWritable(sum));
  }
}

客户端代码：

Job job = Job.getInstance(conf, "Top 10 Sales");
job.setJarByClass(Top10Sales.class);

job.setMapperClass(SalesMapper.class);
job.setReducerClass(SalesReducer.class);

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);

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

job.waitForCompletion(true);

// 读取输出结果，按照销量进行排序
Map<String, Integer> salesMap = new HashMap<>();
FileSystem fs = FileSystem.get(conf);
Path outputPath = new Path(args[1]);
FileStatus[] fileStatuses = fs.listStatus(outputPath);
for (FileStatus fileStatus : fileStatuses) {
  if (fileStatus.isFile()) {
    Path path = fileStatus.getPath();
    BufferedReader reader = new BufferedReader(new InputStreamReader(fs.open(path)));
    String line = reader.readLine();
    while (line != null) {
      String[] fields = line.split("\\s+");
      String productName = fields[0];
      int sales = Integer.parseInt(fields[1]);
      salesMap.put(productName, sales);
      line = reader.readLine();
    }
    reader.close();
  }
}

// 按照销量进行排序，取出销量前十的商品
List<Map.Entry<String, Integer>> salesList = new ArrayList<>(salesMap.entrySet());
Collections.sort(salesList, (o1, o2) -> o2.getValue() - o1.getValue());
for (int i = 0; i < 10 && i < salesList.size(); i++) {
  Map.Entry<String, Integer> entry = salesList.get(i);
  System.out.println(entry.getKey() + "\t" + entry.getValue());
}