MapReduceJob

在Hadoop中，每个MapReduce任务都被初始化为一个Job。

每个Job又可以分为两个阶段：Map阶段和Reduce阶段。这两个阶段分别用Map函数和Reduce函数来表示。

Map函数接收一个<key,value>形式的输入，然后产生另一种<key,value>的中间输出，Hadoop负责将所有具有相同中间key值的value集合到一起传递给Reduce函数；Reduce函数接收一个如<key,(list of values)>形式的输入，然后对这个value集合进行处理并输出结果，Reduce的输出也是<key,value>形式的。

InputFormat()和InputSplit

InputSplit是Hadoop中用来把输入数据传送给每个单独的Map，InputSplit存储的并非数据本身，而是起始位置、分片长度和一个记录数据所在主机的数组。生成InputSplit的方法可以通过InputFormat()来设置。当数据传送给Map时，Map会将输入分片传送到InputFormat()上，InputFormat调用getRecordReader()方法生成RecordReader，RecordReader在通过creatKey()、createValue()方法将InputSplit创建成可供Map处理的<key,value>对。即，InputFormat()方法是用来生成可供Map处理的<key,value>对的。

InputFormat

BaileyBorweinPlouffe.BbpInputFormat

ComposableInputFormat

CompositeInputFormat

DBInputFormat

DistSum.Machine.AbstractInputFormat

FileInputFormat

CombineFileInputFormat

KeyValueTextInputFormat

NLineInputFormat

SequenceFileInputFormat

TeraInputFormat

TextInputFormat

TextInputFormat是Hadoop默认的输入方式。在TextInputFormat中，每个文件（或其一部分）都会单独作为Map的输入，而这是继承自FileInputFormat的。之后，每行数据都会生成一条记录，每条记录则表示成<key,value>形式。

key值是每个数据记录在数据分片中的字节偏移量，数据类型是LongWritable；

value值是每行的内容，数据类型是Text。

如：

file1:

0 hello world bye world

file2:

0 hello hadoop bye hadoop

两个文件都会被单独输入到一个Map中，因此它们的值都是0。

OutputFormat()

默认的输出格式是TextOutputFormat，每条记录以一行的形式存入文本文件，键和值是任意形式的，程序内部调用toString()方法将键和值转换为String类型再输出。

Map函数和Reduce函数：

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{protected void map(LongWritable key, Text value,Context context)throws IOException, InterruptedException {String[] words = StringUtils.split(value.toString(), ' ');for(String w :words){context.write(new Text(w), new IntWritable(1));}}
}

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{protected void reduce(Text key, Iterable<IntWritable> values,Context context)throws IOException, InterruptedException {int sum =0;for(IntWritable i: values){sum=sum+i.get();}arg2.write(key, new IntWritable(sum));}
}

public class RunJob {public static void main(String[] args) {Configuration config =new Configuration();
//      config.set("fs.defaultFS", "hdfs://node1:8020");
//      config.set("yarn.resourcemanager.hostname", "node1");config.set("mapred.jar", "C:\\Users\\Administrator\\Desktop\\wc.jar");try {FileSystem fs =FileSystem.get(config);Job job =Job.getInstance(config);job.setJarByClass(RunJob.class);job.setJobName("wc");job.setMapperClass(WordCountMapper.class);job.setReducerClass(WordCountReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job, new Path("/usr/input/"));Path outpath =new Path("/usr/output/wc");if(fs.exists(outpath)){fs.delete(outpath, true);}FileOutputFormat.setOutputPath(job, outpath);boolean f= job.waitForCompletion(true);if(f){System.out.println("job completion");}} catch (Exception e) {e.printStackTrace();}}
}