java - 如何在java mapreduce hadoop中获得两个键的最大计数

Question

我有一个包含 6 列的 txt 文件，我对第三和第四列、城市和产品感兴趣，这是一个示例:

2015-01-01;09:00:00;New York;shoes;214.05;Amex >

我需要按城市获取销量最大的产品。我已经有了按城市聚合和计算所有产品的代码，这是类映射器和类缩减器的代码:

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class ContaMaxCidadeProdutoMapper extends Mapper<Object, Text, Text, IntWritable> {

	private final static Text cidadeproduto = new Text();
	private final static IntWritable numeroum = new IntWritable(1);

	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
		
		String[] linha=value.toString().split(";");		
		cidadeproduto.set(linha[2] +" "+linha[3]);
		context.write(cidadeproduto, numeroum);		
	}
}

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

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

按城市获取每个产品的计数工作正常，但现在我需要按城市获取最大计数的产品。我知道如何获取整个数据集的最大计数乘积，但我不知道如何按城市获取。我将不胜感激任何提示!谢谢

Answer 1

您想获得城市数量最多的产品。在我看来，您希望每个城市都有产品，并且在该特定城市的销售额最大，不是吗？

我宁愿用 2 个 M-R 对来做。第一对与你的相似:

public void map(Object key, Text value, Context context) {
    String[] linha = value.toString().split(";");       
    cidadeproduto.set(linha[2] + "&" + linha[3]);
    context.write(cidadeproduto, new IntWritable(1));       
}

public void reduce(Text key, Iterable<IntWritable> values, Context context){
    int contValue = 0;

    for (IntWritable value : values) {
        contValue += value.get();
    }
    context.write(key, new IntWritable(contValue));
}

还有第二对。
映射器将重新组合您的数据，使城市成为关键，产品和数量成为值(value):

public void map(Object key, Text value, Context context) {
    String[] row = value.toString().split(";");
    String city = row[0].split("&")[0];
    String product = row[0].split("&")[1];
    String count = row[1];
    context.write(new Text(city), new Text(product + "&" + count));     
}

然后reduce会为每个城市保持最大值:

public void reduce(Text key, Iterable<Text> values, Context context){
    int maxVal = Integer.MIN_VALUE;
    String maxProd = "None";

    for (IntWritable value : values) {
        String ss = value.toString().split("&");
        int cnt = Integer.parseInt(ss[1]);
        if(cnt > maxVal){
            maxVal = cnt;
            maxProd = ss[0];
        }
    }
    context.write(key, new Text(maxProd));
}