源码：

（源码来自网络）

package testDemo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;
import java.io.IOException;

/**
 * @Auther: lp
 * @Date: 2018/10/16 16:04
 * @Description:
 */
public class Phone_Drive {
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(Phone_Drive.class);
        job.setMapperClass(Phone_Map.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        job.setNumReduceTasks(3);

        Path path = setPath(job);

        path.getFileSystem(conf).delete(path,true);//删除你的输出文件夹,可以不用写，但测试时候每次都要删除输出路径文件
        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }

    /**
     * 这个方法为定义数据的输入输出路径
     * @param job
     * @return Path
     * @throws IOException
     */
    private static Path setPath(Job job) throws IOException {
        FileInputFormat.addInputPath(job, new Path("hdfs:/Initial_Data/jdn*/"));
        Path path = new Path("hdfs:/Clean_Data/");
        FileOutputFormat.setOutputPath(job,path);
        return path;
    }

}

分析笔记：

private static Path setPath(Job job)

函数用来设置输入输出路径；
参数为Job类型的job；
利用FileInputFormat.addInputPath()函数添加要读取的文件的路径；
利用FileOutputFormat.setOutputPath()函数设置清洗后的数据保存的路径；

public static void main(String[] args)

main函数，程序的入口；

1. 获取job对象
2. 设置jar存储位置
3. 关联mapper和reducer类
4. 设置mapper阶段输出的key和value的类型
5. 设置最终输出的key和value的类型
6. 设置输入输出的路径
7. 查看结果

修改后的代码：

package testDemo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;
import java.io.IOException;

/**
 * @Auther: lp
 * @Date: 2018/10/16 16:04
 * @Description:
 *  
 * @Modified by: 乐乐今天吃桃子
 * @Date:2019/10/24
 * @Description:增加了关联reducer类的语句；
 * 增加了设置mapper阶段输出的kv类型的语句；
 * 修改了设置输入输出路径的方式；
 * 增加了部分注释
 */
public class Phone_Drive {
    public static void main(String[] args) throws Exception {
       //1.获取job对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
		
		//2.设置jar存储位置
        job.setJarByClass(Phone_Drive.class);
        
		//3.关联mapper和reducer类
		job.setMapperClass(Phone_Map.class);
		job.setReducerClass(Phone_Reducer.class);

		//4.设置mapper阶段输出的key和value的类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);

		//5.设置最终输出的key和value的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

		//6.设置输入输出的路径
        //job.setNumReduceTasks(3);不知道是干啥的，好像是设置Reducer数量的，注释掉
		//因为感觉程序本身不长，写一个函数专门去设置路径会降低可读性，所以就写一起了，所以整个Driver类就只剩下一个函数了。
		FileInputFormat.addInputPath(job, new Path("hdfs:/Initial_Data/jdn*/"));
        Path path = new Path("hdfs:/Clean_Data/");
        FileOutputFormat.setOutputPath(job,path);
        //Path path = setPath(job);
        path.getFileSystem(conf).delete(path,true);//删除你的输出文件夹,可以不用写，但测试时候每次都要删除输出路径文件

		//7.查看结果，如果成功code为0，失败为1；
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}