Mapreduce

是什么

是Hadoop中的分布式计算框架

• 优点：

  1. 易于编程： MR将所有的计算抽象为Map(映射) 与Reduce(聚合) 两个阶段 只需要继承并实现Mapper和Reducer类，就可以完成高性能的分布式程序

  2. 扩展性 与HDFS类似，HDFS是通过将多台机器的存储能力整合到集群中，提供更大的存储能力，MR是通过将多台机器的计算能力(cpu、内存)，提供海量数据的计算

  3. 高容错 高并发(多线程)的分布式程序运行过程中，一些线程出现错误或者某些机器出现故障时，MR框架可以自动启动错误重试机制，或将任务转移到其他机器运行，可以保证任务最终正确执行

  4. 适合处理超大规模数据 MR不适合处理小数据量级，而随着数据量级增大，HDFS可以存储的数据量级，MR都可以使用相同的应用程序完成计算

• 缺点：

  1. 计算延迟较高，不是和实时计算场景

  2. MR任务启动时，需要读取已经存储在磁盘中的文件，如果文件不断动态追加，则MR任务无法启动，所以不能处理流式计算场景

  3. MR任务表达能力有限，一个MR只能完成一次映射和一次聚合，DAG任务如果需要多次聚合，则需要将任务拆分成多个MR，每个MR任务都需要进行大量的磁盘IO，导致性能低下

核心思想

1.分布式的运算程序往往需要分成至少2个阶段。

2.第一个阶段的maptask并发实例，完全并行运行，互不相干。

3.第二个阶段的reduce task并发实例互不相干，但是他们的数据依赖于上一个阶段的所有maptask并发实例的输出。

4.MapReduce编程模型只能包含一个map阶段和一个reduce阶段，如果用户的业务逻辑非常复杂，那就只能多个mapreduce程序，串行运行。

编程规范

用户编写的程序分成三个部分：

Mapper，Reducer，Driver(提交运行mr程序的客户端)

1.Mapper阶段

用户自定义的Mapper要继承自己的父类

Mapper的输入数据是KV对的形式（KV的类型可自定义）

Mapper中的业务逻辑写在map()方法中

Mapper的输出数据是KV对的形式（KV的类型可自定义）

map()方法（maptask进程）对每一个调用一次

2.Reducer阶段

用户自定义的Reducer要继承自己的父类

Reducer的输入数据类型对应Mapper的输出数据类型，也是KV

Reducer的业务逻辑写在reduce()方法中

Reducetask进程对每一组相同k的组调用一次reduce()方法

3.Driver阶段

整个程序需要一个Drvier来进行提交，提交的是一个描述了各种必要信息的job对象

编程模型

单词计数（面试题）

 

• 通常一个典型的MR程序 需要实现三个类

  • Mapper

    1. 自定义一个类 继承Mapper，填写输入输出kv的四个泛型

    2. Mapper包含四个方法 setup(context) 在map任务执行前 执行一次 map(KEYIN k,VALUEIN v,context) 每次获取一组输入的kv对，进行处理，并将处理完的结果交给context进行写出 cleanup(context) 在map任务执行后 执行一次 run() 将上面三个方法组织起来执行Mapper的逻辑

 1 package com.jwl.mappereduce.countWords;
 2 
 3 import org.apache.hadoop.io.IntWritable;
 4 import org.apache.hadoop.io.LongWritable;
 5 import org.apache.hadoop.io.Text;
 6 import org.apache.hadoop.mapreduce.Mapper;
 7 
 8 import java.io.IOException;
 9 
10 public class Job_WordCountMapper
11     // Mapper有四个泛型
12     //  分别是  Mapper输入的k和v类型  以及 输出的k v
13     //   KEYIN, VALUEIN, KEYOUT, VALUEOUT
14     //   如果读取文本文件，则默认输入的K是LongWritable
15     //当前行在文本中的开始位置(字节偏移量offset)
16     // V是 Text 是当前行文件的内容
17     //   Mapper处理完的数据   <单词,1>
18     // 行字节偏移量   行内容  单词   1
19     extends Mapper<LongWritable, Text, Text, IntWritable> {
20 
21   Text k = new Text();
22   IntWritable v = new IntWritable(1);
23 
24   @Override
25   protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
26    // 1. 将读取到的文本的每行数据 切分成单词
27     String[] words = value.toString().split("\\s+");
28   // 2. 将单词进行处理  转小写，去掉特殊符号
29     for (String word : words) {
30       String w = word.toLowerCase()
31           .replaceAll("\\W", "");
32       // 3. 将单词作为当前输出的k值
33       k.set(w);
34 
35 // 4. 使用上下文对象 context.write()
36 // 将Map处理完的结果( <单词,1> ) 写出到MR框架
37       context.write(k, v);
38     }
39   }
40 }

• Reducer

1. 自定义一个类 继承Reducer，填写输入输出kv的四个泛型

2. 与Mapper类似也有4个方法 reduce(KEYIN k, Iterable values)方法每次接收一个key和相同Key对应的所有Value 在reduce方法中对数据进行聚合 并将处理完的结果交给context进行写出

 1 package com.jwl.mappereduce.countWords;
 2 
 3 import org.apache.hadoop.io.IntWritable;
 4 import org.apache.hadoop.io.LongWritable;
 5 import org.apache.hadoop.io.Text;
 6 import org.apache.hadoop.mapreduce.Reducer;
 7 public class Job_WordCountReducer
 8     // Reducer与Mapper类似也有4个泛型
 9     // mapper输出的kv类型,   单词   数量
10     extends Reducer<Text, IntWritable, Text, LongWritable> {
11   @Override
12   protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
13     // 声明变量用于存储聚合完的结果
14     long count = 0;
15     // 遍历相同Key对应的所有value
16       // 对数量进行累加
17       Iterator<IntWritable> iterator = values.iterator();
18  while (iterator.hasNext()){
19    count+=iterator.next().get();
20         }
21     //使用context.write()将reducer聚合完的结果输出到MR框架
22     context.write(key, new LongWritable(count));
23   }
24 }

• Driver 是一个包含main方法的MR任务的入口类 main中 获取job对象实例并添加各种配置 提交job到集群运行

 1 package com.jwl.mappereduce.countWords;
 2 
 3 import org.apache.hadoop.fs.Path;
 4 import org.apache.hadoop.io.IntWritable;
 5 import org.apache.hadoop.io.LongWritable;
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.mapreduce.Job;
 8 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 9 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
10 
11 import java.io.IOException;
12 
13 public class Job_WordCountDriver {
14   public static void main(String[] args) throws Exception {
15     // 0. 如果执行MR任务时需要设置自定义配置，可以使用conf对象
16     Configuration conf = new Configuration();
17 //    conf.set("mapreduce.input.fileinputformat.split.minsize","111");
18     // 1. 创建Job对象实例
19     Job job = Job.getInstance();
20     // 2. 给job对象添加driver类的class
21     job.setJarByClass(Job_WordCountDriver.class);
22     // 3. 给job对象添加mapper类的class
23     job.setMapperClass(Job_WordCountMapper.class);
24     // 4. 给job对象添加reducer类的class
25     job.setReducerClass(Job_WordCountReducer.class);
26 
27     // 5. 设置Mapper输出数据的Key的类型
28    job.setMapOutputKeyClass(Text.class);
29     // 6. 设置Mapper输出数据的Value的类型
30     job.setMapOutputValueClass(IntWritable.class);
31 
32     // 7. 设置Reducer输出数据的Key的类型
33     job.setOutputKeyClass(Text.class);
34    // 8. 设置Reducer输出数据的Value的类型
35     job.setOutputValueClass(LongWritable.class);
36 
37     // 9. 设置MR任务的输入路径
38    FileInputFormat.setInputPaths(job,new Path("C:\\Users\\Administrator\\Desktop\\aa.txt"));
39     // 10. 设置MR任务的输出路径
40      FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\Administrator\\Desktop\\output"));
41 
42     // 11. 执行集群
43     boolean b = job.waitForCompletion(true);
44     System.exit(b ? 0 : 1);
45 
46   }
47 }

通过Java代码实现方法：

 1 package com.jwl.mappereduce;
 2 
 3 import org.apache.commons.io.FileUtils;
 4 
 5 import java.io.File;
 6 import java.io.IOException;
 7 import java.util.*;
 8 
 9 public class CountWords {
10     public static void main(String[] args) throws IOException {
11         //创建存储集合
12         HashMap counts=new HashMap();
13         //获得读取文件
14         File file=new File("C:\\Users\\Administrator\\Desktop\\练习\\大数据\\12-21 mapreduce\\aa.txt");
15         //读取文件
16         List lines= FileUtils.readLines(file,"utf-8");
17         //System.out.println(lines);
18         //解析数据
19    for (String line:lines) {
20  //将数据进行拆分（\\s+空格、\\W特殊字符）
21 String[] words = line.split("\\s+");
22        for (String word:words) {
23                 String s = word.toLowerCase().replaceAll("\\W","");
24                 counts.put(s,counts.getOrDefault(s,0)+1);
25             }
26         }
27         //输出
28        // System.out.println(counts);
29         //排序
30         ArrayList> entries=new ArrayList<>(counts.entrySet());
31        /* entries.sort(new Comparator>() {
32             @Override
33             public int compare(Map.Entry o1, Map.Entry o2) {
34                 return o2.getValue()-o1.getValue();
35             }
36         });*/
37       
38 entries.sort(((o1, o2) -> o1.getValue()-o2.getValue()));
39 for (Map.Entry en:entries) {
40   //正则表达式
41 System.out.printf("单词是%s：出现的个数是%d\n",en.getKey(),en.getValue());
42         }
43     }
44 }

程序运行流程

1.在MapReduce程序读取文件的输入目录上存放相应的文件。

2.客户端程序在submit()方法执行前，获取待处理的数据信息，然后根据集群中参数的配置形成一个任务分配规划。

3.客户端提交job.split、jar包、job.xml等文件给yarn，yarn中的resourcemanager启动MRAppMaster。

4.MRAppMaster启动后根据本次job的描述信息，计算出需要的maptask实例数量，然后向集群申请机器启动相应数量的maptask进程。

5.maptask利用客户指定的inputformat来读取数据，形成输入KV对。

6.maptask将输入KV对传递给客户定义的map()方法，做逻辑运算

7.map()运算完毕后将KV对收集到maptask缓存。

8.maptask缓存中的KV对按照K分区排序后不断写到磁盘文件

9.MRAppMaster监控到所有maptask进程任务完成之后，会根据客户指定的参数启动相应数量的reducetask进程，并告知reducetask进程要处理的数据分区。

10.Reducetask进程启动之后，根据MRAppMaster告知的待处理数据所在位置，从若干台maptask运行所在机器上获取到若干个maptask输出结果文件，并在本地进行重新归并排序，然后按照相同key的KV为一个组，调用客户定义的reduce()方法进行逻辑运算。

11.Reducetask运算完毕后，调用客户指定的outputformat将结果数据输出到外部存储。

• 例1：将多个小文件合并成一个文件

 1 package com.jwl.mappereduce.inputformat;
 2 
 3 import org.apache.hadoop.io.Text;
 4 import org.apache.hadoop.mapreduce.InputSplit;
 5 import org.apache.hadoop.mapreduce.RecordReader;
 6 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 7 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 8 
 9 import java.io.IOException;
10 
11 public class AllInputformat extends FileInputFormat {
12     /**
13      *
14      * @param inputSplit      输入切片
15      * @param taskAttemptContext  上下文对象
16      * @return    获得的自定义读取器
17      * @throws IOException
18      * @throws InterruptedException
19      */
20     @Override
21     public RecordReader createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
22         //获取读取器对象
23         AllRecordReader allRecordReader=new AllRecordReader();
24         //初始化读取器
25 allRecordReader.initialize(inputSplit,taskAttemptContext);
26         return allRecordReader;
27     }
28 }

 1 package com.jwl.mappereduce.inputformat;
 2 
 3 import org.apache.hadoop.conf.Configuration;
 4 import org.apache.hadoop.fs.FSDataInputStream;
 5 import org.apache.hadoop.fs.FileStatus;
 6 import org.apache.hadoop.fs.FileSystem;
 7 import org.apache.hadoop.fs.Path;
 8 import org.apache.hadoop.io.Text;
 9 
10 import org.apache.hadoop.mapreduce.InputSplit;
11 import org.apache.hadoop.mapreduce.RecordReader;
12 import org.apache.hadoop.mapreduce.TaskAttemptContext;
13 import org.apache.hadoop.mapreduce.lib.input.FileSplit;
14 
15 import java.io.IOException;
16 
17 public class AllRecordReader extends RecordReader {
18     FileSplit fileSplit=null;
19     FileSystem fs=null;
20     Text k=new Text();
21     Text v=new Text();
22     boolean falg=true;
23     @Override
24     public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
25         //获得配置文件对象
26         Configuration configuration=new Configuration();
27         //获得文件系统
28 fs=FileSystem.get(configuration);
29         //将输入的切片转换成文件切片
30 fileSplit=(FileSplit)inputSplit;
31     }
32 
33     @Override
34     public boolean nextKeyValue() throws IOException, InterruptedException {
35         if (falg) {
36             //获得文件路径
37 Path path = fileSplit.getPath();
38             //获得输入流
39             FSDataInputStream inputStream = fs.open(path);
40             //根据路径获得文件状态
41             FileStatus[] fileStatuses = fs.listStatus(path);
42             FileStatus fileStatus1 = fileStatuses[0];
43             String name = fileStatus1.getPath().getName();
44             //获得长度
45 long len = fileStatus1.getLen();
46             //创建缓冲区
47             byte[] data = new byte[(int) len];
48             //读取数据
49             inputStream.read(data);
50             inputStream.close();
51         this.k.set(name);
52 this.v.set(new String(data));
53             falg=false;
54             return true;
55         }
56         return false;
57     }
58 
59     @Override
60     public Text getCurrentKey() throws IOException, InterruptedException {
61         return this.k;
62     }
63 
64     @Override
65     public Text getCurrentValue() throws IOException, InterruptedException {
66         return this.v;
67     }
68 
69     @Override
70     public float getProgress() throws IOException, InterruptedException {
71         return falg?0.0F:1.0F;
72     }
73 
74     @Override
75     public void close() throws IOException {
76         this.fs.close();
77     }
78 }

 1 package com.jwl.mappereduce.inputformat;
 2 
 3 
 4 import org.apache.hadoop.fs.Path;
 5 
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.mapreduce.Job;
 8 import org.apache.hadoop.mapreduce.Mapper;
 9 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
10 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
11 
12 import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
13 
14 import java.io.IOException;
15 
16 public class AllinputformatDriver {
17     public static class AllinputformatMapper extends Mapper{
18         @Override
19         protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
20             context.write(key,value);
21         }
22     }
23     public static void main(String[] args) throws Exception {
24         Job job=Job.getInstance();
25 job.setJarByClass(AllinputformatDriver.class);
26 job.setMapperClass(AllinputformatMapper.class);
27 job.setMapOutputKeyClass(Text.class);
28 job.setMapOutputValueClass(Text.class);
29 job.setOutputKeyClass(Text.class);
30 job.setOutputValueClass(Text.class);
31         //设值自定义输入格式
32 job.setInputFormatClass(AllInputformat.class);
33 job.setOutputFormatClass(SequenceFileOutputFormat.class);
34 FileInputFormat.setInputPaths(job,new Path("C:\\Users\\Administrator\\Desktop\\12"));
35 FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\Administrator\\Desktop\\11"));
36 boolean b = job.waitForCompletion(true);
37         System.exit(b?0:1);
38     }
39 }

• 例2：将合成的文件ip进行统计

 1 package com.jwl.mappereduce.inputformat;
 2 
 3 import org.apache.hadoop.fs.Path;
 4 import org.apache.hadoop.io.IntWritable;
 5 import org.apache.hadoop.io.LongWritable;
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.mapreduce.Job;
 8 import org.apache.hadoop.mapreduce.Mapper;
 9 import org.apache.hadoop.mapreduce.Reducer;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
12 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
13 import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
14 
15 import java.io.IOException;
16 
17 public class Job_IpCountDriver {
18     public static class Job_IpCountMapper
19             extends Mapper{
20         Text k = new Text();
21 IntWritable v = new IntWritable(1);
22         @Override
23         protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
24             //根据行进行拆分
25             String[] lines = value.toString().split("\n");
26           for (String line:lines ) {
27     //将读取数据按照空格截取成字符数组
28                 String[] split = line.toString().split(" ");
29          k.set(split[0]);//绑定key为ip
30                 context.write(k,v);
31             }
32         }
33     }
34     //reduce
35 public static class Job_IpCountReduce
36             extends Reducer {
37 
38         @Override
39         //  a      [1,1,1,1,1]
40         protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
41             int count =0;
42    for (IntWritable value:values) {
43                 count+=value.get();
44             }
45             context.write(key,new LongWritable(count));
46         }
47     }
48     public static void main(String[] args) throws Exception{
49         Job job = Job.getInstance();
50 job.setJarByClass(Job_IpCountDriver.class);
51 job.setMapperClass(Job_IpCountMapper.class);
52 job.setReducerClass(Job_IpCountReduce.class);
53 job.setMapOutputKeyClass(Text.class);
54 job.setMapOutputValueClass(IntWritable.class);
55 job.setOutputKeyClass(Text.class);
56 job.setOutputValueClass(LongWritable.class);
57         //指定输入的格式是序列化文件
58 job.setInputFormatClass(SequenceFileInputFormat.class);
59 FileInputFormat.setInputPaths(job,new Path("C:\\Users\\Administrator\\Desktop\\11"));
60 FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\Administrator\\Desktop\\out34"));
61 boolean b = job.waitForCompletion(true);
62         System.exit(b ? 0 : 1);
63     }
64 }

• 按照ip首字母设置分区

 1 package com.jwl.mappereduce.point;
 2 
 3 import org.apache.hadoop.io.LongWritable;
 4 import org.apache.hadoop.io.Text;
 5 import org.apache.hadoop.mapreduce.Partitioner;
 6 
 7 public class IpPartitioner extends Partitioner {
 8     @Override
 9     public int getPartition(Text text, LongWritable longWritable, int numPartitions) {
10       //将ip首字母提取并转换成int类型
11 int i =text.toString().charAt(0)-'0';
12   //通过首字母对设置的reduce个数进行取余分区
13       return i%numPartitions;
14     }
15 }

 1 package com.jwl.mappereduce.point;
 2 
 3 import org.apache.hadoop.fs.Path;
 4 import org.apache.hadoop.io.LongWritable;
 5 import org.apache.hadoop.io.Text;
 6 import org.apache.hadoop.mapreduce.Job;
 7 import org.apache.hadoop.mapreduce.Mapper;
 8 import org.apache.hadoop.mapreduce.Reducer;
 9 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
10 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
11 
12 import java.io.IOException;
13 
14 public class Job_IpCountDriver {
15     public static class Job_IpCountMapper extends Mapper{
16         Text k = new Text();
17 LongWritable v = new LongWritable(1);
18 
19         @Override
20         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
21             //根据行进行拆分
22             String[] lines = value.toString().split("\n");
23 //将读取数据按照空格截取成字符数组
24                 String[] split = value.toString().split(" ");
25        k.set(split[0]);//绑定key为ip
26           context.write(k,v);
27 
28         }
29         //reduce
30         public static class Job_IpCountReduce extends Reducer {
31 
32             @Override
33             protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
34                 int count =0;
35  for (LongWritable value:values) {
36  count+=value.get();
37                 }
38                 context.write(key,new LongWritable(count));
39             }
40         }
41         public static void main(String[] args) throws Exception{
42 Job job = Job.getInstance();
43 job.setJarByClass(Job_IpCountDriver.class);
44 job.setMapperClass(Job_IpCountMapper.class);
45 job.setReducerClass(Job_IpCountReduce.class);
46 job.setMapOutputKeyClass(Text.class);
47 job.setMapOutputValueClass(LongWritable.class);
48 job.setOutputKeyClass(Text.class);
49 job.setOutputValueClass(LongWritable.class);
50             //分区方式
51 job.setPartitionerClass(IpPartitioner.class);
52             //设值reduce个数
53 job.setNumReduceTasks(11);
54 FileInputFormat.setInputPaths(job,new Path("C:\\Users\\Administrator\\Desktop\\bb.txt"));
55 FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\Administrator\\Desktop\\inputOut4"));
56  boolean b = job.waitForCompletion(true);
57             System.exit(b ? 0 : 1);
58         }
59     }
60 }

1. 数据输入

• MR程序启动任务时，会使用FileInputFormat计算任务的分片数

  1. FileInputFormat默认使用Block大小作为分片大小Math.max(minSize, Math.min(maxSize, blockSize));

  2. 除此之外，FileInputFormat设置了一个阈值1.1,如果文件大小不超过分片大小的1.1倍，则直接作为一个分片

  3. 如果需要调整分片大小，则可以通过修改以下配置 想减小分片大小,修改mapreduce.input.fileinputformat.split.maxsize 想增加分片大小,修改mapreduce.input.fileinputformat.split.minsize

• 分片的个数与MR任务启动的MapTask的线程数(并发度)对应

• 特殊场景下，由于默认TextInputFormat每个文件至少产生一个分片，如果原始数据是大量小文件会导致启动过多MapTask线程导致性能收到影响

  • MR中提供了可以跨文件进行分片合并的输入格式化类，CombineTextInpuFormat

1 // driver中设置使用的输入格式化类
2 job.setInputFormatClass(CombineTextInputFormat.class);
3 // 设置CombineTextInputFormat的最大最小分片大小，通常设置为一样值，可以完成跨文件的分片输入
4 CombineTextInputFormat.setMinInputSplitSize(job, 100 * 1024 * 1024);
5 CombineTextInputFormat.setMaxInputSplitSize(job, 100 * 1024 * 1024);

• MR也允许开发者根据自己的需要创建自定义的数据输入格式化类

  1. 自定义一个类 继承 FileInputFormat,实现createRecordReader()方法

  2. 自定义一个类 继承RecordReader,实现6个方法

1 init() 初始化方法用于获取分片信息和上下文对象
2 nextKeyValue() 判断是否还有下一组kv，以及读取数据将数据赋值给当前的k v
3 getCurKey() 返回当前K
4 getCurValue() 返回当前V
5 getProgress() 返回读取进度
6 close() 释放资源

1. 在Driver中使用job.setInputFormatClass(设置自定义输入格式化类)

2. MapTask阶段

• Mapper

  • setup 在map前获取一些资源或者连接

  • map(k,v,context) 编写处理逻辑，使用context写出处理后的kv

  • cleanup 在map后关闭资源释放连接

  • run 将上述三个方法组织成外部调用的公共方法

• Mapper和Reducer如果需要处理包含多个属性的负责对象时，Hadoop自身提供的Writable不能满足需要，可以自定义个实现Wirtable序列化的bean

  1. 声明一个类实现Wirtable接口

  2. 实现序列化方法write(out)

  3. 实现反序列化方法readFields(ip)

  4. 重写toString方便数据的输出 注意: 序列化和反序列化的顺序必须一致

3. 环形缓冲区

• MR提供了一个默认100M大小内存区域，用于存储Map输出的kv

• 当缓冲区到达存储空间0.8的阈值时，将已经存储在缓冲区的数据刷写到磁盘中

• 由于使用了环形设计，可以保证数据写入和读取互不影响，使用顺序读写

4. 分区

• MR在Map结束后Reduce开始前 会对数据按照Key进行分区

• 默认分区数据量为1，如果job中设置了reduceTask数量，则分区数量与reduceTask数量一致

  • MR默认使用HashPartitioner，使用Key的hash % reduceTask，HashParitioner可以比较均衡的将Map输出的输入分配给每一个reducer避免数据倾斜

  • 如果需要将数据按照要求输出到不同的文件，可以自定义一个类继承Partitioner，自己实现getPartition方法用于计算分区编号。自定义分区器可能导致数据倾斜，可以提前将较大Key单独处理或者继续散列为多个key

• 分区数reduce数量的关系

  • 分区数 = reduce数 理想状态

  • 分区数 > reduce数 任务报错

  • 分区数 < reduce数 生成多余空文件

5. 排序

• MR任务会默认按照Map输出数据的Key进行升序排列

• 排序时会使用Key位置上的对象的compareTo方法进行升序排列

• 如果需要让自定义的Writable类放在输出Key位置上，则需要实现WritableComparable接口，并实现compareTo()方法用于进行对象比较

6. Map端预聚合

• Combiner是MR提供的一个Map端的预聚合机制

• 在Map输出之后，没有产生网络shuffle之前，在Map端对数据进行类似Reducer的聚合操作

• 编写MR任务时，可以直接将Reducer作为Combiner

• 观察如果没有改变计算结果，则可以使用预聚合优化

7. reduce端合并

• 所有的MapTask结束后，会根据分区的编号

• 每个Reduce读取一个分区(来自多个MapTask)的数据，进行合并

• 合并时 将相同K对应的所有Value合并到一个集合中

例3：对stu和sc文件进行信息和成绩合并

 1 package com.jwl.mappereduce.join;
 2 
 3 import org.apache.hadoop.fs.Path;
 4 import org.apache.hadoop.io.LongWritable;
 5 import org.apache.hadoop.io.Text;
 6 import org.apache.hadoop.mapreduce.InputSplit;
 7 import org.apache.hadoop.mapreduce.Job;
 8 import org.apache.hadoop.mapreduce.Mapper;
 9 import org.apache.hadoop.mapreduce.Reducer;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.input.FileSplit;
12 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
13 
14 import java.io.IOException;
15 import java.util.Iterator;
16 import java.util.LinkedList;
17 import java.util.List;
18 
19 public class Join_Driver {
20     public static class Join_Mapper extends Mapper {
21 // k  01   v  101 赵雷 1990-01-01 男
22         // k  01   v  001 01 80
23         Text k = new Text();
24         Text v = new Text();
25 
26         @Override
27         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
28             //获得文件名
29             InputSplit inputSplit = context.getInputSplit();
30             FileSplit fileSplit = (FileSplit) inputSplit;
31             String name = fileSplit.getPath().getName();
32             String[] split = value.toString().split(" ");
33             //根据文件名区分值
34 if (name.startsWith("stu")) {
35                 v.set("1" + value);
36             } else {
37                 v.set("0" + value);
38             }
39             k.set(split[0]);
40             context.write(k, v);
41         }
42     }
43 
44     public static class Join_Reduce extends Reducer {
45         @Override
46         protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
47              /*01   [
48                 101 赵雷
49                 001 01 80
50                 ]
51 
52            k 01    v 01 赵雷 01 02                                 80*/
53             String student = null;
54 List score = new LinkedList<>();
55             Iterator iterator = values.iterator();
56 while (iterator.hasNext()) {
57                 String next = iterator.next().toString();
58 String index = next.substring(0, 1);
59  if ("1".equals(index)) {
60 student = next.substring(1);
61                 } else {
62 score.add(next.substring(1));
63                 }
64             }
65             //根据成绩进行循环拼接
66             for (String sc : score) {
67 Text v = new Text(student + " " + sc);
68           context.write(key, v);
69             }
70         }
71     }
72 
73     public static void main(String[] args) throws Exception {
74         Job job = Job.getInstance();
75 job.setJarByClass(Join_Driver.class);
76 job.setMapperClass(Join_Mapper.class);
77 job.setReducerClass(Join_Reduce.class);
78 job.setMapOutputKeyClass(Text.class);
79 job.setMapOutputValueClass(Text.class);
80 job.setOutputKeyClass(Text.class);
81 job.setOutputValueClass(Text.class);
82 FileInputFormat.setInputPaths(job, new Path("C:\\Users\\Administrator\\Desktop\\aa"));
83  FileOutputFormat.setOutputPath(job, new Path("C:\\Users\\Administrator\\Desktop\\out2"));
84 boolean b = job.waitForCompletion(true);
85         System.exit(b ? 0 : 1);
86     }
87 }

8. ReduceTask阶段

• setup 创建连接 申请资源

• reduce(K key, Iterable values,context )是ReduceTask的核心方法，用于完成聚合的逻辑

• claenup 关闭资源

• run 将上述三个方法组合执行reduce任务

9. 数据输出

• MR任务默认使用TextOutputFormat，将K和V toString后使用\t进行分割 每个KV对输出到一行中

• 可以在driver中job.setOuputFormat(xxx.class)改变输出格式化

• 也可以根据需要自定义输出格式化

  1. 自定义类继承FileOutputFormat,实现方法createRecordWirter

  2. 自定义类继承RecordWirter,实现方法 write(K key,V value) 拿到reducer输出后的kv对，可以在wirte方法中实现io或者jdbc将数据写出到对应的自定义文件或者数据库 close()关闭连接或者流

例1：根据ip地址获得城市名称和ip并将数据输出成文本类型

 1 package com.jwl.mappereduce.utils;
 2 
 3 import org.apache.http.HttpEntity;
 4 import org.apache.http.client.methods.CloseableHttpResponse;
 5 import org.apache.http.client.methods.HttpGet;
 6 import org.apache.http.impl.client.CloseableHttpClient;
 7 import org.apache.http.impl.client.HttpClients;
 8 import org.apache.http.util.EntityUtils;
 9 
10 import java.io.IOException;
11 import java.util.Arrays;
12 import java.util.List;
13 
14 public class IPGEOUtils {
15     public static void main(String[] args) throws IOException {
16         //获得城市ip地址
17         String ip = "207.46.13.72";
18         //创建get方法
19         getIpGEO(ip);
20     }
21 
22     //  创建一个http客户端对象
23     static CloseableHttpClient client = HttpClients.createDefault();
24     //网址路径
25     static String url = "http://ip.ws.126.net/ipquery?ip=";
26                              
27     public static List getIpGEO(String ip) throws IOException {
28         //获得get   http://ip.ws.126.net/ipquery?ip=207.46.13.72
29  HttpGet get = new HttpGet(url + ip);
30         if (client == null) {
31 client = HttpClients.createDefault();
32         }
33         //执行get请求并获得响应
34         CloseableHttpResponse response = client.execute(get);
35         //获得响应实体对象
36         HttpEntity entity = response.getEntity();
37         //将对象转成字符串
38         //var lo="华盛顿州", lc="雷德蒙德"; var localAddress={city:"雷德蒙德", province:"华盛顿州"}
39         String s = EntityUtils.toString(entity);
40         //根据""对字符串进行拆分
41 String[] split = s.split("\"");
42         //将第一位拆分获得华盛顿州
43         String pro = split[1];
44         //将第三位拆分获得雷德蒙德
45         String city = split[3];
46         //华盛顿州:雷德蒙德
47 System.out.println(pro + ":" + city);
48         //根据地址获得所有城市列表
49 return Arrays.asList(pro, city);
50     }
51 }

  1 package com.jwl.mappereduce.outputformat;
  2 
  3 
  4 import com.jwl.mappereduce.utils.IPGEOUtils;
  5 import org.apache.hadoop.conf.Configuration;
  6 import org.apache.hadoop.fs.FSDataOutputStream;
  7 import org.apache.hadoop.fs.FileSystem;
  8 import org.apache.hadoop.fs.Path;
  9 
 10 import org.apache.hadoop.io.LongWritable;
 11 import org.apache.hadoop.io.Text;
 12 import org.apache.hadoop.mapreduce.*;
 13 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 14 
 15 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 16 
 17 import java.io.IOException;
 18 import java.util.HashMap;
 19 import java.util.List;
 20 
 21 public class IpGEODriver {
 22     public static class IpGEOMapper extends Mapper {
 23         Text k = new Text();
 24         Text v = new Text();
 25 
 26         @Override
 27         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
 28             //var lo="华盛顿州", lc="雷德蒙德"; var localAddress={city:"雷德蒙德", province:"华盛顿州"}
 29 //拆分第0位获得ip：207.46.13.72
 30 String ip = value.toString().split(" ")[0];
 31             //获得ip的城市
 32             List ipGEO = IPGEOUtils.getIpGEO(ip);
 33      //根据集合获得第一位：雷德蒙德
 34   String city = ipGEO.get(1);
 35             k.set(city + ":" + ip);
 36             //上下文写入键和值
 37             context.write(k, v);
 38         }
 39     }
 40 
 41     public static class IpGEOReduce extends Reducer {
 42   Text k = new Text();//雷德蒙德
 43   Text v = new Text();//207.46.13.72
 44 
 45         @Override
 46         protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
 47             //雷德蒙德:123.5.7.6
 48             String[] split = key.toString().split(":");
 49             k.set(split[0]);
 50             v.set(split[1]);
 51             context.write(k, v);
 52         }
 53     }
 54 
 55     public static void main(String[] args) throws Exception {
 56         //获得配置
 57         Configuration conf = new Configuration();
 58         //创建输出路径
 59         conf.set("ipUrl", "C:\\Users\\Administrator\\Desktop\\out");
 60 Job job = Job.getInstance(conf);
 61         //获得jar包
 62 job.setJarByClass(IpGEODriver.class);
 63 job.setMapperClass(IpGEOMapper.class); job.setReducerClass(IpGEOReduce.class);
 64 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class);
 65 job.setOutputKeyClass(Text.class);
 66 job.setOutputValueClass(Text.class);
 67         //设值格式化输出
 68 job.setOutputFormatClass(CityOutputformat.class);
 69 FileInputFormat.setInputPaths(job, new Path("C:\\Users\\Administrator\\Desktop\\city.txt"));
 70         FileOutputFormat.setOutputPath(job, new Path("C:\\Users\\Administrator\\Desktop\\out"));
 71 
 72         boolean b = job.waitForCompletion(true);
 73         System.exit(b ? 0 : 1);
 74     }
 75 
 76     //自定义一个FileOutputfaormat
 77     public static class CityOutputformat extends FileOutputFormat {
 78 
 79         @Override
 80     public RecordWriter getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
 81 return new GEOREcordWriter(job);
 82         }
 83     }
 84 
 85     public static class GEOREcordWriter extends RecordWriter {
 86      FileSystem fileSystem = null;
 87         HashMap outputStream = new HashMap();
 88 
 89         public GEOREcordWriter(TaskAttemptContext context) {
 90 Configuration configuration = context.getConfiguration();
 91             try {
 92                 fileSystem = FileSystem.get(configuration);
 93           } catch (IOException e) {
 94                 e.printStackTrace();
 95             }
 96         }
 97 
 98         @Override
 99         public void write(Text key, Text value) throws IOException, InterruptedException {
100    String city = key.toString();
101             //确定存放的位置
102             Configuration conf = fileSystem.getConf();
103 String parent = conf.get("ipUrl");
104             //获得对外输出流
105             Path path = new Path(parent + "/" + city + "地区的用户.txt");
106 if (!outputStream.containsKey(city)) {
107                 FSDataOutputStream fsDataOutputStream = fileSystem.create(path);
108 outputStream.put(city, fsDataOutputStream);
109             }
110             //向外输出信息
111             String data = city + "\t" + value.toString();
112             outputStream.get(city).write(data.getBytes());
113         }
114 
115         @Override
116         public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
117             for (FSDataOutputStream fs : outputStream.values()) {
118                 fs.flush();//冲刷
119                 fs.close();//关闭
120             }
121         }
122     }
123 }

计数器

• 由于在MR程序中，分布式运行过程中会启动若干个Map/ReduceTask(线程)，如果希望统计整个任务的某些环节的执行次数或者数据量，目的是为了在优化程序后可以对比优化成果

• 如果自己手动实现计数，需要考虑将多个线程的计算结果合并，编码过于麻烦

• MR提供了两种方式直接创建MR程序全局计数器

  • context.getCounter("组名","计数器名称")

  • context.getCounter( Enum.枚举值 )

• 使用Counter.incriment()进行累加操作

例2：对文件ip和以“127”开头的ip进行计数

 1 package com.jwl.mappereduce.count;
 2 
 3 import org.apache.hadoop.fs.Path;
 4 import org.apache.hadoop.io.IntWritable;
 5 import org.apache.hadoop.io.LongWritable;
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.mapreduce.Counter;
 8 import org.apache.hadoop.mapreduce.Job;
 9 import org.apache.hadoop.mapreduce.Mapper;
10 import org.apache.hadoop.mapreduce.Reducer;
11 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
12 import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
13 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
14 
15 import java.io.IOException;
16 
17 public class Job_IpCountDriver {
18     public static class Job_IpCountMapper
19             extends Mapper {
20         Text k = new Text();
21 IntWritable v = new IntWritable(1);
22 
23         @Override
24         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
25 
26        //将读取数据按照空格截取成字符数组
27             String[] split = value.toString().split(" ");
28      k.set(split[0]);//绑定key为ip
29        context.write(k,v);
30 
31     //计数器
32     //方式1：通过枚举类型获取ip总和
33                 Counter counter = context.getCounter(EnumCounter.allCount_Ip);
34    //方式2：通过字符串获得127开头的ip
35                 Counter counter1 = context.getCounter("自定义组件", "127开头的ip");
36 if (split[0].startsWith("127")){
37                     //自增1
38 counter1.increment(1);
39                 }
40       counter.increment(1);
41         }
42     }
43 
44     //reduce
45     public static class Job_IpCountReduce extends Reducer {
46 
47         @Override
48         // a           [1,1,1,1,1]
49         protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
50             int count = 0;
51    for (IntWritable value : values) {
52                 count += value.get();
53             }
54             context.write(key, new LongWritable(count));
55         }
56     }
57 
58     public static void main(String[] args) throws Exception {
59         Job job = Job.getInstance();
60 job.setJarByClass(Job_IpCountDriver.class);
61 job.setMapperClass(Job_IpCountMapper.class);
62 job.setReducerClass(Job_IpCountReduce.class);
63 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class);
64 job.setOutputKeyClass(Text.class);
65 job.setOutputValueClass(LongWritable.class);
66         FileInputFormat.setInputPaths(job, new Path("C:\\Users\\Administrator\\Desktop\\log(1).log"));
67         FileOutputFormat.setOutputPath(job, new Path("C:\\Users\\Administrator\\Desktop\\out1"));
68 
69         boolean b = job.waitForCompletion(true);
70         System.exit(b ? 0 : 1);
71     }
72 }

1 package com.jwl.mappereduce.count;
2 
3 public enum EnumCounter {
4     start_127,allCount_Ip
5 }

Hadoop序列化

• 常见序列化类型

• 自定义对象实现序列化接口

1.必须实现Writable接口

2.反序列化时，需要反射调用空参构造函数，所以必须有空参构造

3.重写序列化方法

4.重写反序列化方法

5.注意反序列化的顺序和序列化的顺序完全一致

6.要想把结果显示在文件中，需要重写toString()，且用”\t”分开，方便后续用

7.如果需要将自定义的bean放在key中传输，则还需要实现comparable接口，因为mapreduce框中的shuffle过程一定会对key进行排序

统计每个IP的访问次数和访问流量

 1 package com.jwl.mappereduce.ip;
 2 
 3 import org.apache.hadoop.fs.Path;
 4 import org.apache.hadoop.io.IntWritable;
 5 import org.apache.hadoop.io.LongWritable;
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.mapreduce.Job;
 8 import org.apache.hadoop.mapreduce.Mapper;
 9 import org.apache.hadoop.mapreduce.Reducer;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12 
13 import java.io.IOException;
14 
15 public class Job_IpCountDriver {
16     public static class Job_IpCountMapper extends Mapper{
17         Text k=new Text();
18 IntWritable v=new IntWritable(1);
19 
20         @Override
21         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
22   //将读取数据按照空格截取成字符数组
23             String[] s = value.toString().split(" ");
24       k.set(s[0]);//绑定key为ip
25             context.write(k,v);
26         }
27     }
28     public static class Job_IpCountReduce extends Reducer{
29         @Override
30         protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
31             int count =0;
32   for (IntWritable value:values) {
33                 count+=value.get();
34             }
35   context.write(key,new LongWritable(count));
36         }
37     }
38 
39     public static void main(String[] args) throws Exception {
40         Job job=Job.getInstance();
41 job.setJarByClass(Job_IpCountDriver.class);
42 job.setMapperClass(Job_IpCountMapper.class);
43 job.setReducerClass(Job_IpCountReduce.class);
44 job.setMapOutputKeyClass(Text.class);
45 job.setMapOutputValueClass(IntWritable.class);
46 job.setOutputKeyClass(Text.class);
47 job.setOutputValueClass(LongWritable.class);
48 FileInputFormat.setInputPaths(job,new Path("C:\\Users\\Administrator\\Desktop\\log(1).log"));
49 FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\Administrator\\Desktop\\inputOut"));
50 
51  boolean b = job.waitForCompletion(true);
52         System.exit(b?0:1);
53     }
54 }

序列化排序：

  1 package com.jwl.mappereduce.entity;
  2 
  3 import org.apache.hadoop.io.Writable;
  4 import org.apache.hadoop.io.WritableComparable;
  5 
  6 import java.io.DataInput;
  7 import java.io.DataOutput;
  8 import java.io.IOException;
  9 
 10 public class IpDetail implements WritableComparable {
 11     private String ip;
 12     private int IpCount;
 13     private int getCount;
 14     private int postCount;
 15 
 16     public String getIp() {
 17         return ip;
 18     }
 19 
 20     public void setIp(String ip) {
 21         this.ip = ip;
 22     }
 23 
 24     public int getIpCount() {
 25         return IpCount;
 26     }
 27 
 28     public void setIpCount(int ipCount) {
 29         IpCount = ipCount;
 30     }
 31 
 32     public int getGetCount() {
 33         return getCount;
 34     }
 35 
 36     public void setGetCount(int getCount) {
 37         this.getCount = getCount;
 38     }
 39 
 40     public int getPostCount() {
 41         return postCount;
 42     }
 43 
 44     public void setPostCount(int postCount) {
 45         this.postCount = postCount;
 46     }
 47 
 48     public IpDetail() {
 49     }
 50 
 51     public IpDetail(String ip, int ipCount, int getCount, int postCount) {
 52         this.ip = ip;
 53         IpCount = ipCount;
 54         this.getCount = getCount;
 55         this.postCount = postCount;
 56     }
 57     public void set(String ip, int ipCount, int getCount, int postCount) {
 58         this.ip = ip;
 59         IpCount = ipCount;
 60         this.getCount = getCount;
 61         this.postCount = postCount;
 62     }
 63 
 64     @Override
 65     public void write(DataOutput dataOutput) throws IOException {
 66         //序列化
 67    dataOutput.writeUTF(this.ip);
 68 dataOutput.writeInt(this.IpCount);
 69 dataOutput.writeInt(this.getCount);
 70 dataOutput.writeInt(this.postCount);
 71     }
 72     @Override
 73     public void readFields(DataInput dataInput) throws IOException {
 74         //反序列化
 75 this.ip = dataInput.readUTF();
 76 this.IpCount=dataInput.readInt();
 77 this.getCount=dataInput.readInt();
 78 this.postCount=dataInput.readInt();
 79     }
 80 
 81     public void add(IpDetail ipD) {
 82         this.IpCount+=ipD.IpCount;
 83         this.getCount+=ipD.getCount;
 84   this.postCount+=ipD.postCount;
 85     }
 86 
 87     @Override
 88     public String toString() {
 89         return  ip + "\t" +
 90                 IpCount +"\t" +
 91                 getCount +"\t" +
 92                 postCount;
 93     }
 94 
 95     @Override
 96   public int compareTo(IpDetail o) {
 97 if (this.getCount==o.getCount){
 98             return Long.compare(this.IpCount,o.IpCount);
 99         }
100         return Long.compare(this.getCount,o.getCount);
101     }
102 }

对文件中ip、ip合计、get合计、post合计进行统计

 1 package com.jwl.mappereduce.ip;
 2 
 3 import com.jwl.mappereduce.entity.IpDetail;
 4 import org.apache.hadoop.fs.Path;
 5 import org.apache.hadoop.io.LongWritable;
 6 import org.apache.hadoop.io.Text;
 7 import org.apache.hadoop.mapreduce.Job;
 8 import org.apache.hadoop.mapreduce.Mapper;
 9 import org.apache.hadoop.mapreduce.Reducer;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12 
13 import java.io.IOException;
14 
15 public class Job_IpDetailDriver {
16 //ip    ip  访问次数  get次数  post次数
17 //127.0.0.1     127.0.0.1 10      8       2
18     public static class Job_IpDetailMapper extends Mapper{
19         @Override
20         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
21             String[] split = value.toString().split(" ");
22             Text k =new Text();
23             k.set(split[0]);
24 int getCount="GET".equals(split[3])?1:0;
25 int postCount="POST".equals(split[3])?1:0;
26             IpDetail v=new IpDetail(split[0],1,getCount,postCount);
27             context.write(k,v);
28         }
29     }
30 public static class Job_IpDetailReduce extends Reducer{
31         @Override
32         protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
33 IpDetail sum = new IpDetail();
34  sum.setIp(key.toString());
35      for (IpDetail ipD:values) {
36                 sum.add(ipD);//聚合
37             }
38             context.write(key,sum);
39         }
40     }
41 
42     public static void main(String[] args) throws Exception{
43         Job job = Job.getInstance();
44 job.setJarByClass(Job_IpDetailDriver.class);
45 job.setMapperClass(Job_IpDetailMapper.class);
46 job.setReducerClass(Job_IpDetailReduce.class);
47 job.setMapOutputKeyClass(Text.class);
48 job.setMapOutputValueClass(IpDetail.class);
49 job.setOutputKeyClass(Text.class);
50 job.setOutputValueClass(IpDetail.class);
51 FileInputFormat.setInputPaths(job,new Path("C:\\Users\\Administrator\\Desktop\\log(1).log"));
52 FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\Administrator\\Desktop\\out1"));
53 
54         boolean b = job.waitForCompletion(true);
55         System.exit(b ? 0 : 1);
56     }
57 }

将文件ip合计次数和ip进行交换

 1 package com.jwl.mappereduce.ip;
 2 
 3 import com.jwl.mappereduce.entity.IpDetail;
 4 import org.apache.hadoop.fs.Path;
 5 import org.apache.hadoop.io.LongWritable;
 6 import org.apache.hadoop.io.NullWritable;
 7 import org.apache.hadoop.io.Text;
 8 import org.apache.hadoop.mapreduce.Job;
 9 import org.apache.hadoop.mapreduce.Mapper;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12 
13 import java.io.IOException;
14 
15 public class Job_IpCountSortDriver {
16     public static class Job_IpCountSortMapper extends Mapper{
17         @Override
18         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
19             String[] split = value.toString().split("\t");
20             LongWritable k=new LongWritable(Long.parseLong(split[1]));
21             Text v=new Text(split[0]);
22             context.write(k,v);
23         }
24     }
25     public static void main(String[] args) throws Exception {
26         Job job=Job.getInstance();
27 job.setJarByClass(Job_IpCountSortDriver.class);
28 job.setMapperClass(Job_IpCountSortMapper.class);
29 job.setMapOutputKeyClass(LongWritable.class);
30 job.setMapOutputValueClass(Text.class);
31 FileInputFormat.setInputPaths(job,new Path("C:\\Users\\Administrator\\Desktop\\inputOut\\part-r-00000"));
32 FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\Administrator\\Desktop\\inputOut2"));
33         boolean b = job.waitForCompletion(true);
34         System.exit(b?0:1);
35     }
36 }

将文件get次数按照升序排列，每当遇到get次数相同是按照ip合计次数升序排列

 1 package com.jwl.mappereduce.ip;
 2 import com.jwl.mappereduce.entity.IpDetail;
 3 import org.apache.hadoop.fs.Path;
 4 import org.apache.hadoop.io.IntWritable;
 5 import org.apache.hadoop.io.LongWritable;
 6 import org.apache.hadoop.io.NullWritable;
 7 import org.apache.hadoop.io.Text;
 8 import org.apache.hadoop.mapreduce.Job;
 9 import org.apache.hadoop.mapreduce.Mapper;
10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
11 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
12 
13 import java.io.IOException;
14 
15 public class Job_IpDetailSortDriver {
16    public static class Job_IpDetailSortMapper extends Mapper{
17        IpDetail k = new IpDetail();
18        @Override
19        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
20            String[] split = value.toString().split("\t");
21            String ip=split[0];
22            int ipCount=Integer.parseInt(split[2]);
23            int getCount=Integer.parseInt(split[3]);
24            int PostCount=Integer.parseInt(split[4]);
25 k.set(ip,ipCount,getCount,PostCount);
26 context.write(k,NullWritable.get());
27        }
28    }
29     public static void main(String[] args) throws Exception {
30         Job job=Job.getInstance();
31 job.setJarByClass(Job_IpDetailSortDriver.class);
32 job.setMapperClass(Job_IpDetailSortMapper.class);
33 job.setMapOutputKeyClass(IpDetail.class);
34 job.setMapOutputValueClass(NullWritable.class);
35 FileInputFormat.setInputPaths(job,new Path("C:\\Users\\Administrator\\Desktop\\out1\\part-r-00000"));
36 FileOutputFormat.setOutputPath(job,new Path("C:\\Users\\Administrator\\Desktop\\outof3"));
37 boolean b = job.waitForCompletion(true);
38         System.exit(b?0:1);
39     }
40 }