使用Hadoop Map Reduce执行多个计算

Question

我有一个map reduce程序，用于查找每年两个独立属性的最小值/最大值。在大多数情况下，这是通过在hadoop中使用单个节点集群来实现的。这是我目前的设置：

public class MaxTemperatureReducer extends
        Reducer<Text, Stats, Text, Stats> {
    private Stats result = new Stats();
    @Override
    public void reduce(Text key, Iterable<Stats> values, Context context)
            throws IOException, InterruptedException {
        int maxValue = Integer.MIN_VALUE;
        int minValue = Integer.MAX_VALUE;
        int sum = 0;
        for (Stats value : values) {
            result.setMaxTemp(Math.max(maxValue, value.getMaxTemp()));
            result.setMinTemp(Math.min(minValue, value.getMinTemp()));
            result.setMaxWind(Math.max(maxValue, value.getMaxWind()));
            result.setMinWind(Math.min(minValue, value.getMinWind()));
            sum += value.getCount();
        }
        result.setCount(sum);
        context.write(key, result);
    }
}
public class MaxTemperatureMapper extends
        Mapper<Object, Text, Text, Stats> {
    private static final int MISSING = 9999;
    private Stats outStat = new Stats();
    @Override
    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] split = value.toString().split("\s+");
        String year = split[2].substring(0, 4);
        int airTemperature;
        airTemperature = (int) Float.parseFloat(split[3]);
        outStat.setMinTemp((float)airTemperature);
        outStat.setMaxTemp((float)airTemperature);
        outStat.setMinWind(Float.parseFloat(split[12]));
        outStat.setMaxWind(Float.parseFloat(split[14]));
        outStat.setCount(1);
        context.write(new Text(year), outStat);
    }
}
public class MaxTemperatureDriver extends Configured implements Tool {
    public int run(String[] args) throws Exception {
        if (args.length != 2) {
            System.err
                    .println("Usage: MaxTemperatureDriver <input path> <outputpath>");
            System.exit(-1);
        }
        Job job = new Job();
        job.setJarByClass(MaxTemperatureDriver.class);
        job.setJobName("Max Temperature");
        job.setMapperClass(MaxTemperatureMapper.class);
        job.setCombinerClass(MaxTemperatureReducer.class);
        job.setReducerClass(MaxTemperatureReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Stats.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
        boolean success = job.waitForCompletion(true);
        return success ? 0 : 1;
    }
    public static void main(String[] args) throws Exception {
        MaxTemperatureDriver driver = new MaxTemperatureDriver();
        int exitCode = ToolRunner.run(driver, args);
        System.exit(exitCode);
    }
 }

目前，它只打印每年的温度和风速的最小值/最大值。我确信这是一个简单的实现，但在任何地方都找不到答案。我想试着找到每年最好的5分钟/最大值。有什么建议吗？

Answer 1

让我假设您的Stats类具有以下签名。

/* the stats class need to be a writable, the below is just a demo*/
public class Stats {
public float getTemp() {
    return temp;
}
public void setTemp(float temp) {
    this.temp = temp;
}
public float getWind() {
    return wind;
}
public void setWind(float wind) {
    this.wind = wind;
}
private float temp;
private float wind;
}

有了这个，让我们按如下方式更换减速器。

SortedSet<Float> tempSetMax = new TreeSet<Float>();
        SortedSet<Float> tempSetMin = new TreeSet<Float>();
        SortedSet<Float> windSetMin = new TreeSet<Float>();
        SortedSet<Float> windSetMax = new TreeSet<Float>();
        List<Stats> values = new ArrayList<Float>();
        for (Stats value : values) {
            float temp = value.getTemp();
            float wind = value.getWind();
            if (tempSetMax.size() < 5) {
                tempSetMax.add(temp);
            } else {
                float currentMinValue = tempSetMax.first();
                if (temp > currentMinValue) {
                    tempSetMax.remove(currentMinValue);
                    tempSetMax.add(temp);
                }
            }
            if (tempSetMin.size() < 5) {
                tempSetMin.add(temp);
            } else {
                float currentMaxValue = tempSetMin.last();
                if (temp < currentMaxValue) {
                    tempSetMax.remove(currentMaxValue);
                    tempSetMax.add(temp);
                }
            }
            if (windSetMin.size() < 5) {
                windSetMin.add(wind);
            } else {
                float currentMaxValue = windSetMin.last();
                if (wind < currentMaxValue) {
                    windSetMin.remove(currentMaxValue);
                    windSetMin.add(temp);
                }
            }
            if (windSetMax.size() < 5) {
                windSetMax.add(wind);
            } else {
                float currentMinValue = windSetMax.first();
                if (wind > currentMinValue) {
                    windSetMax.remove(currentMinValue);
                    windSetMax.add(temp);
                }
            }
        }

现在，您可以编写每个列表的toString()的上下文，也可以创建一个自定义的可写。在我的代码中，请根据您的要求更改Stats。它需要是一个Writable。以上内容仅用于演示示例流程。

Answer 2

以下是MR设计模式书中获得前10名的代码。在同一个GitHub位置还有其他MR设计模式的代码。