流量统计之Driver开发（第四步）

package mr.access;

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

public class AccessLocalApp {

    // Driver端的代码：八股文式的模板
    public static void main(String[] args) throws Exception{

        //1、创建Configuration
        Configuration configuration = new Configuration();

        //2、创建一个Job
        //将configuration传进来
        Job job = Job.getInstance(configuration);

        //3、设置Job对应的参数：主类
        job.setJarByClass(AccessLocalApp.class);

        //4、设置Job对应的参数：设置自定义的Mapper和Reducer处理类
        job.setMapperClass(AccessMapper.class);
        job.setReducerClass(AccessReducer.class);

        //5、添加Combiner的设置即可
        //job.setCombinerClass(AccessReducer.class);

        //6、设置Job对应的参数：Mapper输出key和value的类型
        //不需要关注Mapper输入
        //Mapper<LongWritable, Text, Text, Access>
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Access.class);

        //7、设置Job对应的参数：Reducer输出key和value的类型
        //不需要关注Reducer输入
        //Reducer<Text, Access, Text, Access>
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Access.class);

        //设置Job对应的参数：Mapper输出key和value的类型：作业输入和输出的路径
        FileInputFormat.setInputPaths(job, new Path("access/input"));
        FileOutputFormat.setOutputPath(job, new Path("access/output"));

        //提交job
        boolean result = job.waitForCompletion(true);

        System.exit(result ? 0 : -1);
    }
}