import java.io.IOException;

//导入IO异常类，当发生某种 I/O 异常时，抛出此异常。此类是失败或中断的 I/O 操作生成的异常的通用类。

public class IOException extends Exception

import java.util.Iterator;

//导入接口Iterator，Iterator是对 collection 进行迭代的迭代器。迭代器取代了 Java Collections Framework 中的 Enumeration。迭代器与枚举有两点不同：迭代器允许调用者利用定义良好的语义在迭代期间从迭代器所指向的 collection 移除元素。方法名称得到了改进。  public interface Iterator

//Collection是最基本的集合接口，一个Collection代表一组Object，即Collection的元素(Elements)。一些Collection允许相同的元素而另一些不行。一些能排序而另一些不行。Java SDK不提供直接继承自Collection的类，Java SDK提供的类都是继承自Collection的"子接口"如List和Set。

import org.apache.hadoop.conf.Configuration;

//hadoop使用了一套独有的配置文件管理系统，并提供自己的API，即使用org.apache.hadoop.conf.Configuration处理配置信息。详细参看《hadoop的API  Configuration》

//是MapReduce配置模块最底层的类，支持序列化（实现Iterator接口）和迭代（实现Writable接口）

public class Configuration

extends Object

implements Iterable<Map.Entry<String,String>>, Writable

import org.apache.hadoop.conf.Configured;

// org.apache.hadoop.conf中的最后一个类，也是这个包中以后用的最频繁的一个，Configurable算是肉体，Configuration算是灵魂吧

// Configurable是一个很简单的接口，也位于org.apache.hadoop.conf包中, 从字面解释，Configurable的含义是可配置的，如果一个类实现了Configurable接口，意味着这个类是可配置的。也就是说，可以通过为这个类的对象传一个Configuration实例，提供对象工作需要的一些配置信息。在setConf()中，如果对象实现了Configurable接口，那么对象的setConf()方法会被调用，并根据Configuration类的实例conf进一步初始化对象。

public interface Configurable

public class Configured

extends Object

implements Configurable

import org.apache.hadoop.fs.Path;

// Path    对路径进行解析，将参数转换为标准的URI格式，对Path的参数作判断，标准化，字符化等操作。

public class Path  extends Object   implements  Comparable

//Names a file or directory in a FileSystem. Path strings use slash as the directory separator. A path string is absolute if it begins with a slash.

import org.apache.hadoop.io.Text;

// This class stores text using standard UTF8 encoding. It provides methods to serialize, deserialize, and compare texts at byte level. The type of length is integer and is serialized using zero-compressed format.

public class Text

extends BinaryComparable

implements WritableComparable<BinaryComparable>

import org.apache.hadoop.mapred.FileInputFormat;

//InputFormat接口的实现

//为Map-Reduce 作业描述输入的细节规范。1、检查作业输入的有效性。2、把输入文件切分成多个逻辑InputSplit实例， 并把每一实例分别分发给一个 Mapper。3、提供RecordReader的实现，这个RecordReader从逻辑InputSplit中获得输入记录， 这些记录将由Mapper处理。

public abstract class FileInputFormat<K,V>

extends Object

implements InputFormat<K,V>

import org.apache.hadoop.mapred.FileOutputFormat;

//OutputFormat接口的实现

//描述Map-Reduce 作业的输出样式。1、检验作业的输出，例如检查输出路径是否已经存在。2、提供一个RecordWriter的实现，用来输出作业结果。 输出文件保存在FileSystem上。

public abstract class FileOutputFormat<K,V>

extends Object

implements OutputFormat<K,V>

import org.apache.hadoop.mapred.JobClient;

//JobClient是用户提交的作业与JobTracker交互的主要接口, 提供提交作业， 追踪进程，访问子任务的日志记录， 获得Map-reduce集群状态信息等功能。

public class JobClient

extends Configured

implements Tool

import org.apache.hadoop.mapred.JobConf;

//代表一个Map-Reduce作业的配置，描述一个map-reduce作业运行时所需要的所有信息。

//JobConf会自动加载配置文件mapred-default.xml 和 mapred-site.xml

public class JobConf

extends Configuration

import org.apache.hadoop.mapred.KeyValueTextInputFormat;

// KeyValueTextInputFormat：用于读取文件，如果行被分隔符分割为两部分，第一部分为key，剩下的为value；若没有分隔符，整行作为key，value为空。

public class KeyValueTextInputFormat

extends FileInputFormat<Text,Text>

implements JobConfigurable

import org.apache.hadoop.mapred.MapReduceBase;

//Mapper和Reducer的基类，提供了的close()和configure(JobConf job)方法，继承子类需要编写两个方法的实现。

public class MapReduceBase

extends Object

implements Closeable, JobConfigurable

import org.apache.hadoop.mapred.Mapper;

//将输入键值对(key/value pair)映射到一组中间格式的 键值对集合。

public interface Mapper<K1,V1,K2,V2>

extends JobConfigurable, Closeable

import org.apache.hadoop.mapred.OutputCollector;

// Map端的上下文， OutputCollector和Reporter是Hadoop-0.19以前版本里面的API，在Hadoop-0.20.2以后就换成Context，Context的功能包含了OutputCollector和Reporter的功能。OutputCollector 由 Hadoop 框架提供, 负责收集 Mapper 和 Reducer 的输出数据，实现 map 函数和 reduce 函数时，只需要简单地将其输出的 <key,value> 对往 OutputCollector 中一丢即可，剩余的事框架自会帮你处理好。

public interface OutputCollector<K,V>

import org.apache.hadoop.mapred.Reducer;

//reducer,将与一个key关联的一组中间数值集归约（reduce）为一个更小的数值集。

public interface Reducer<K2,V2,K3,V3>

extends JobConfigurable, Closeable

import org.apache.hadoop.mapred.Reporter;

//Reporter用于Map-Reduce应用程序报告进度，设定应用级别的状态消息， 更新Counters（计数器）的机制。

public interface Reporter

extends Progressable

import org.apache.hadoop.mapred.TextOutputFormat;

// TextInputFormat：用于读取纯文本文件，文件被分为一系列以LF或CR结束的行，key是每一行的偏移量(LongWritable)，value是每一行的内容(Text)。

//TextOutputFormat：用于输出一个纯文本文件。将每个记录写为一行文本。键和值以字符串的形式写入，并以制表符（\t）分隔。

public class TextOutputFormat<K,V>

extends FileOutputFormat<K,V>

import org.apache.hadoop.util.Tool;

// Tool接口可以支持处理通用的命令行选项，它是所有Map-Reduce程序的都可用的一个标准接口

public interface Tool

extends Configurable

int	run(String[] args)           Execute the command with the given arguments.

import org.apache.hadoop.util.ToolRunner;

//定义的一个类，实现Tool接口，在main()方法中通过ToolRunner.run(...)方法调用上述类的run(String[]方法）

public class ToolRunner

extends Object

//定义MyJob类

public class MyJob extends Configured implements Tool {

//实现 Mapper 接口

public static class MapClass extends MapReduceBase

implements Mapper<Text, Text, Text, Text> {

public void map(Text key, Text value,

OutputCollector<Text, Text> output,

Reporter reporter) throws IOException {

output.collect(value, key);

}

}

//实现Reducer接口

public static class Reduce extends MapReduceBase

implements Reducer<Text, Text, Text, Text> {

//输入参数中的 value 是文本文件中的一行

public void reduce(Text key, Iterator<Text> values,

OutputCollector<Text, Text> output,

Reporter reporter) throws IOException {

String csv = "";

// values.hasNext() 如果仍有元素可以迭代，则返回 true。

while (values.hasNext()) {

if (csv.length() > 0) csv += ",";

// values.next() 返回迭代的下一个元素。

csv += values.next().toString();

}

output.collect(key, new Text(csv));

}

}

//重载tool的run方法

public int run(String[] args) throws Exception {

//调用基类Configured的getConf获取环境变量实例

Configuration conf = getConf();

//定义一个map-reduce作业，作业的配置conf，作业被MyJob使用

JobConf job = new JobConf(conf, MyJob.class);

//定义输入路径

Path in = new Path(args[0]);

//定义输出路径

Path out = new Path(args[1]);

//为作业定义输入路径

FileInputFormat.setInputPaths(job, in);

//为作业定义输出路径

FileOutputFormat.setOutputPath(job, out);

//设置作业名称

job.setJobName("MyJob");

//设置作业的分解对象

job.setMapperClass(MapClass.class);

//设置作业的整合对象

job.setReducerClass(Reduce.class);

//设置作业的输入实现，输入的数据用给定的分隔符分割，前面是键key，后面是值value，

//key1、value1 的类型均为text

job.setInputFormat(KeyValueTextInputFormat.class);

//设置作业的输出实现

job.setOutputFormat(TextOutputFormat.class);

//设置作业输出键是纯文本，保证与输入分割时的输出的key的类型一致 key2的类型是text

job.setOutputKeyClass(Text.class);

//设置输出的值是纯文本，保证与输入分割时的输出的value的类型一致 value2的类型是text

job.setOutputValueClass(Text.class);

//设置逗号为输入文本的分割符

job.set("key.value.separator.in.input.line", ",");

//客户端作业类运行作业

JobClient.runJob(job);

return 0;

}

//主方法，args是调用命令行输入的参数数组

public static void main(String[] args) throws Exception {

//调用MyJob的run方法，为MyJob创建一个配置项，传入命令行参数 args

//这里的new MyJob() 作为Tool的实现

//new Configuration，将得到hadoop默认的路径，在core-default.xml中配置

int res = ToolRunner.run(new Configuration(), new MyJob(), args);

System.exit(res);

}

}