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(Text.class);
    job.setMapOutputValueClass(IntWritable.class);
    job.setOutputKeyClass(Text.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, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text time = new Text();

    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[7]);
            // 获取时间
            String timeStr = fields[2];
            if (timeStr.compareTo("12:00:00") >= 0 && timeStr.compareTo("15:00:00") <= 0) {
                time.set(fields[1]);
                context.write(time, new IntWritable(count));
            }
        } catch (NumberFormatException e) {
            return;
        }
    }
}

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

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

        // 计算同一天内12点到3点的使用总数
        for (IntWritable value : values) {
            sum += value.get();
        }

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

}