import java.io.IOException; 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.hado...-Go服务器开发

import java.io.IOException;

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");

        // 判断是否为文件头
        if (fields[0].equals("id")) {
            return;
        }

        // 检查数据格式
        if (fields.length != 11) {
            return;
        }

         // 检查租赁数量是否为数字
        try {
            int count = Integer.parseInt(fields[10]);
        } catch (NumberFormatException e) {
            return;
        }

       // 取整温度
       int temp = (int) Math.floor(Double.parseDouble(fields[6]));

       // 输出<temp, count>
       context.write(new IntWritable(temp), new IntWritable(Integer.parseInt(fields[10])));
    }
}

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));
    }
}

}