java - 在 Java Hadoop 2.2 的 MapReduce 中对一系列值进行分组

Question

我有以下 JSON 格式的输入数据。

        "SeasonTicket": false, 
        "name": "Vinson Foreman", 
        "gender": "male", 
        "age": 50, 
        "email": "vinsonforeman@cyclonica.com", 
        "annualSalary": "$98501.00", 
        "id": 0

我需要根据工资范围对值进行排序，即1000-10000、10000-25000等。

Range        Count 
1000-10000     10
10000-50000    20

我没有使用默认的 JSON 解析器或 Jackson 来处理数据，而是将其解析为字符串。我有以下映射和化简函数。

map 功能

public class DemoMapper extends MapReduceBase
        implements Mapper<LongWritable, Text, Text, IntWritable> {

    private final IntWritable v = new IntWritable(1);
    private Text k = new Text();

    @Override
    public void map(LongWritable key, Text value,
                    OutputCollector<Text, IntWritable> output, Reporter reporter)
            throws IOException {
        try {
            if (line.contains("annualSalary")) {
                String s = line.replaceAll("$", "");
                String t = s.substring(26);
                Double x = Double.parseDouble(t);

                StringTokenizer itr = new StringTokenizer(t);
                while (itr.hasMoreTokens()) {
                    Double x = Double.parseDouble(s.substring(26));
                    if (x > 1000 && x < 10000) {
                        k.set(itr.nextToken());
                        output.collect(k, v);
                    } else if (x > 10000 && x < 50000) {
                        k.set(itr.nextToken());
                        output.collect(k, v);
                    } else if (x > 50000 && x < 100000) {
                        k.set(itr.nextToken());
                        output.collect(k, v);
                    } else if (x > 100000) {
                        k.set(itr.nextToken());
                        output.collect(k, v);
                    }
                    output.collect(k, v);
                }

            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }

    }
}

归约函数

public class DemoReducer extends MapReduceBase
        implements Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable count = new IntWritable();

    @Override
    public void reduce(Text key, Iterator<IntWritable> values,
                       OutputCollector<Text, IntWritable> output, Reporter reporter)
            throws IOException {
        int sum = 0;
        while (values.hasNext()) {
            IntWritable value = (IntWritable) values.next();
            sum += value.get();
        }
        count.set(sum);
        output.collect(key, (IntWritable) count);
    }
}

如果可能，请让我知道在不使用 JSON 解析器的情况下对这些数据进行分组。

Answer 1

要按范围对数据进行分组，您可以使用自定义分区器 - example .