import java.io.IOException; import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.Calendar; import java.util.Date; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.had...-Go服务器开发

import java.io.IOException; import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.Calendar; import java.util.Date;

import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Task1 {

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "task1");
    job.setJarByClass(Task1.class);
    job.setMapperClass(Mapper1.class);
    job.setReducerClass(Reducer1.class);
    // 设置Mapper和Reducer的输出类型
    job.setMapOutputKeyClass(IntWritable.class);
    job.setMapOutputValueClass(IntWritable.class);
    job.setOutputKeyClass(IntWritable.class);
    job.setOutputValueClass(IntWritable.class);

    // 修改输入路径和输出路径
    FileInputFormat.addInputPath(job, new Path("hdfs://localhost:9000/exam/20560640321"));
    FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost:9000/exam/20560640321/output"));

    System.exit(job.waitForCompletion(true) ? 0 : 1);
}

public static class Mapper1 extends Mapper<LongWritable, Text, IntWritable, IntWritable> {
    private final static IntWritable one = new IntWritable(1);

   public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] fields = line.split("\t");
        // 检查使用数量是否为数字
        try {
            int count = Integer.parseInt(fields[10]);
            // 获取温度
            double temperature = Double.parseDouble(fields[4]);
            int roundedTemperature = (int) Math.round(temperature);

            context.write(new IntWritable(roundedTemperature), new IntWritable(count));
            System.out.println("Mapper Output: " + roundedTemperature + "\t" + count);

        } catch (NumberFormatException e) {
            return;
        }
    }
}

public static class Reducer1 extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {

    @Override
    public void reduce(IntWritable key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        int sum = 0;

        // 计算同一温度下的使用总数
        for (IntWritable value : values) {
            sum += value.get();
        }

        // 输出<temperature, total_count>
        context.write(key, new IntWritable(sum));
    }
}

}