package wordcount;
//导入必要的package
import java.io.IOException; //报错类
import java.util.HashSet;
import java.util.Iterator;//迭代器，与string和next有关
import java.util.Set;
import java.util.StringTokenizer; //StringTokenizer类，用于将空白字符作为分割符的类
import org.apache.hadoop.mapreduce.lib.input.FileSplit;//文件分割类，使用它可以获取文件的文件路径等信息。
import org.apache.hadoop.conf.Configuration;//Hadoop中用于读取配置信息的类
import org.apache.hadoop.fs.Path; //有关文件系统输入输出数据的类
import org.apache.hadoop.io.IntWritable; //封装定义了IntWritable类
import org.apache.hadoop.io.Text; //封装定义了Text类
import org.apache.hadoop.mapreduce.Job; //封装定义了Job类
import org.apache.hadoop.mapreduce.Mapper; //封装定义了Mapper类
import org.apache.hadoop.mapreduce.Reducer; //封装定义了Reducer类
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; //文件输入要用到的类
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; //文件输出要用到的类
import org.apache.hadoop.util.GenericOptionsParser; //GenericOptionsParser类，用来解释常用hadoop命令，并根据需要为Configuration对象设置相应的值

public class zhunliyinyong{
public static class TokenizerMapper
extends Mapper<Object, Text, Text, Text>{ //自定义的TokenizerMapper类，继承自前面导入的Mapper类

private Text word = new Text(); //实例化了一个Text类的对象word
private Text word1 = new Text();
public void map(Object key, Text value, Context context //定义Map方法
) throws IOException, InterruptedException {

//这里说一下context类，它是Mapper的一个内部类，它用来与MapReduce系统进行通信，如把map的结果传给reduce处理。简单的说顶级接口用它在map或是reduce任务中跟踪task的状态，MapContext就是记录了map执行的上下文，在mapper类中，这个context可以存储一些job conf的信息，同时context作为了map和reduce执行中各个函数的一个桥梁，我们可以在map函数中处理这个信息
StringTokenizer itr = new StringTokenizer(value.toString());//实例化了一个以空白字符为分隔符的StringTokenizer类的对象itr
while (itr.hasMoreTokens())
{
String[] strs = itr.nextToken().split(",");

   this.word.set(strs[4]);
   this.word1.set(strs[2]);
    context.write(word1, word);  //context.write方法将(value, key)这样的二元组存入context中
  }
}

}

public static class IntSumReducer                           //自定义的IntSumReducer类，继承自前面导入的Reducer类                             
extends Reducer<Text,Text,Text,IntWritable> {
private IntWritable result = new IntWritable();           //实例化了一个IntWritable类的result对象

public void reduce(Text key, Iterable<Text> values,Context context//定义Reduce方法，这里迭代器（Iterator）是一种设计模式，它是一个对象，它可以遍历并选择序列（IntWritable）中的对象，而开发人员不需要了解该序列的底层结构。
                ) throws IOException, InterruptedException {
	Set<String> lala =  new HashSet<String>();;// map的keySet()方法返回类型即Set集合，Set集合的特点是不重复，与map的键不重复是一致的。
	for(Text val:values){
        lala.add(val.toString());
    }   
	int sum=lala.size();
    result.set(sum);
    context.write(key,result);
}

}

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
//运行MapReduce程序前都要初始化Configuration，该类主要是读取MapReduce系统配置信息，这些信息包括hdfs还有MapReduce，也就是安装hadoop时候的配置文件例如：core-site.xml、hdfs-site.xml和mapred-site.xml等等文件里的信息，有些童鞋不理解为啥要这么做，这个是没有深入思考MapReduce计算框架造成，我们程序员开发MapReduce时候只是在填空，在map函数和reduce函数里编写实际进行的业务逻辑，其它的工作都是交给MapReduce框架自己操作的，但是至少我们要告诉它怎么操作啊，比如hdfs在哪里，MapReduce的jobstracker在哪里，而这些信息就在conf包下的配置文件里。

String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length < 2) {
  System.err.println("Usage: wordcount <in> [<in>...] <out>");
  System.exit(2);
}//If的语句好理解，就是运行WordCount程序时候一定是两个参数，如果不是就会报错退出。至于第一句里的GenericOptionsParser类，它是用来解释常用hadoop命令，并根据需要为Configuration对象设置相应的值
Job job = Job.getInstance(conf, "zhunliyinyong");//用Job.getInstance方法设置作业名为word count
job.setJarByClass(zhunliyinyong.class);           //为job的输出数据设置Key类
job.setMapperClass(TokenizerMapper.class);    //设置Mapper类(Map阶段使用)
job.setReducerClass(IntSumReducer.class);     //设置Reducer类（Reduce阶段使用）
job.setOutputKeyClass(Text.class);            //为job的输出数据设置Key类，规定Reduce输出的Key类型为Text
job.setOutputValueClass(Text.class);   //设置Reduce输出的Value类型为Text

for (int i = 0; i < otherArgs.length - 1; ++i) { //设置输入输出路径      
  FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
}
FileOutputFormat.setOutputPath(job,
  new Path(otherArgs[otherArgs.length - 1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);//等待任务执行完毕退出

}
}

package wordcount;
//导入必要的package
import java.io.IOException; //报错类
import java.util.StringTokenizer; //StringTokenizer类，用于将空白字符作为分割符的类

import org.apache.hadoop.conf.Configuration;//Hadoop中用于读取配置信息的类
import org.apache.hadoop.fs.Path; //有关文件系统输入输出数据的类
import org.apache.hadoop.io.IntWritable; //封装定义了IntWritable类
import org.apache.hadoop.io.Text; //封装定义了Text类
import org.apache.hadoop.mapreduce.Job; //封装定义了Job类
import org.apache.hadoop.mapreduce.Mapper; //封装定义了Mapper类
import org.apache.hadoop.mapreduce.Reducer; //封装定义了Reducer类
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; //文件输入要用到的类
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; //文件输出要用到的类
import org.apache.hadoop.util.GenericOptionsParser; //GenericOptionsParser类，用来解释常用hadoop命令，并根据需要为Configuration对象设置相应的值

public class zhuanlicishu{

public static class TokenizerMapper 
extends Mapper<Object, Text, Text, IntWritable>{         //自定义的TokenizerMapper类，继承自前面导入的Mapper类

private final static IntWritable one = new IntWritable(1); //实例化了一个IntWritable类的one对象并赋值为常量1
private Text word = new Text(); //实例化了一个Text类的对象word

public void map(Object key, Text value, Context context //定义Map方法
) throws IOException, InterruptedException {

//这里说一下context类，它是Mapper的一个内部类，它用来与MapReduce系统进行通信，如把map的结果传给reduce处理。简单的说顶级接口用它在map或是reduce任务中跟踪task的状态，MapContext就是记录了map执行的上下文，在mapper类中，这个context可以存储一些job conf的信息，同时context作为了map和reduce执行中各个函数的一个桥梁，我们可以在map函数中处理这个信息
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens())
{
String[] strs = itr.nextToken().split(",");
word.set(strs[4]);
context.write(word, one);
}

}
}
public static class IntSumReducer //自定义的IntSumReducer类，继承自前面导入的Reducer类
extends Reducer<Text,IntWritable,Text,IntWritable> {
private IntWritable result = new IntWritable(); //实例化了一个IntWritable类的result对象

public void reduce(Text key, Iterable values,Context context//定义Reduce方法，这里迭代器（Iterator）是一种设计模式，它是一个对象，它可以遍历并选择序列（IntWritable）中的对象，而开发人员不需要了解该序列的底层结构。
) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();//将该词的出现次数相加
}
result.set(sum);//将sum赋给result
context.write(key, result);//输出最终结果
}
}

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
//运行MapReduce程序前都要初始化Configuration，该类主要是读取MapReduce系统配置信息，这些信息包括hdfs还有MapReduce，也就是安装hadoop时候的配置文件例如：core-site.xml、hdfs-site.xml和mapred-site.xml等等文件里的信息，有些童鞋不理解为啥要这么做，这个是没有深入思考MapReduce计算框架造成，我们程序员开发MapReduce时候只是在填空，在map函数和reduce函数里编写实际进行的业务逻辑，其它的工作都是交给MapReduce框架自己操作的，但是至少我们要告诉它怎么操作啊，比如hdfs在哪里，MapReduce的jobstracker在哪里，而这些信息就在conf包下的配置文件里。

String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length < 2) {
System.err.println("Usage: wordcount […] ");
System.exit(2);
}//If的语句好理解，就是运行WordCount程序时候一定是两个参数，如果不是就会报错退出。至于第一句里的GenericOptionsParser类，它是用来解释常用hadoop命令，并根据需要为Configuration对象设置相应的值
Job job = Job.getInstance(conf, “zhuanlicishu”);//用Job.getInstance方法设置作业名为word count
job.setJarByClass(zhuanlicishu.class); //为job的输出数据设置Key类
job.setMapperClass(TokenizerMapper.class); //设置Mapper类(Map阶段使用)
job.setReducerClass(IntSumReducer.class); //设置Reducer类（Reduce阶段使用）
job.setOutputKeyClass(Text.class); //为job的输出数据设置Key类，规定Reduce输出的Key类型为Text
job.setOutputValueClass(IntWritable.class); //设置Reduce输出的Value类型为IntWritable

for (int i = 0; i < otherArgs.length - 1; ++i) { //设置输入输出路径
FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
}
FileOutputFormat.setOutputPath(job,
new Path(otherArgs[otherArgs.length - 1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);//等待任务执行完毕退出
}
}