大数据-Hadoop-MapReduce

一、MapReduce的概念

Hadoop的三大组件：HDFS、Yarn、MapReduce。
HDFS：解决的是分布式存储的问题。
MapReduce: 解决的是盘算问题。
Yarn: 盘算的时候，使用的资源如何协调（Windows操作系统）

1.     2004年，谷歌发表了一篇名为《MapReduce》的论文，主要介绍了如何在分布式的存储系统上对数据进行高效率的计算。2005年，Nutch团队使用Java语言实现了这个技术，并命名为MapReduce。时至今日，MapReduce是Apache Hadoop的核心模块之一，是运行在HDFS上的分布式运算程序的编程框架，用于大规模数据集（大于1TB）的并行运算。其中的概念，"Map（映射）"和"Reduce（归约）"
3. reduce 其实就是合并的意思。

复制代码

数据量假如太大，盘算是代码随着数据走。

mapReduce的优缺点：

1. 1、易于编程
2.       代码写起来有固定的格式，编写难度非常的小，号称是八股文【固定写法】。
3. 2、良好的扩展性
4.       代码的计算资源不够了，可以直接拓展几台即可解决
5. 3、高容出错
6.       如果负责计算的电脑挂掉了，直接可以将任务转移到其他电脑上，任务不会执行失败的。
7. 4、非常适合大数据集的计算（PB级以上）  1P=1024T

复制代码

1. 1、不适合做实时计算
2.     mapreduce一个任务就要跑很长时间，不利于实时。不能做到秒级或者毫秒级的计算。
3.     mapreduce 属于离线的技术。
4. 2、不适合做流式计算
5.     数据因为都是静态的，不是边产生数据，边计算。
6.     固定计算：数据量是固定的，给了1T 就计算。
7. 3、不适合做有向图（DAG）计算
8.     多个应用程序之间有依赖关系，后一个程序需要依赖前面的程序的结果。这种场景就称之为有向图，mapreduce是不适合的。

复制代码

总结：MapReduce只能做离线的数据分析，并且盘算速度比力慢。
我们之前打仗过：运行过一个WordCount 案例。
二、MapReduce案例--WordCount

1、新建maven项目，并且导入包

1. <!--指定代码编译的版本-->
2. <properties>
3.   <maven.compiler.target>1.8</maven.compiler.target>
4.   <maven.compiler.source>1.8</maven.compiler.source>
5. </properties>
7. <dependencies>
8.   <dependency>
9.     <groupId>org.apache.hadoop</groupId>
10.     <artifactId>hadoop-common</artifactId>
11.     <version>3.3.1</version>
12.   </dependency>
13.   <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client -->
14.   <dependency>
15.     <groupId>org.apache.hadoop</groupId>
16.     <artifactId>hadoop-client</artifactId>
17.     <version>3.3.1</version>
18.   </dependency>
19.   <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-hdfs -->
20.   <dependency>
21.     <groupId>org.apache.hadoop</groupId>
22.     <artifactId>hadoop-hdfs</artifactId>
23.     <version>3.3.1</version>
24.   </dependency>
25. </dependencies>

复制代码

补充Maven的支持：

2、创建一些数据

在项目标根路径下，创建一个文件夹，mr01 ,在mr01文件夹下，再创建数据的来源文件input文件夹，
在input文件夹下面，新建file，a.txt, b.txt, c.txt

1. a.txt
2. hello bigdata hello 1999 hello beijing hello
3. world hello hello java good
5. b.txt
6. hello gaoxinqu hello bingbing
7. hello chenchen hello
8. ACMilan hello china
10. c.txt
11. hello hadoop hello java hello storm hello spark hello redis hello zookeeper
12. hello hive hello hbase hello flume

复制代码

3、编写代码

1）编写Map代码

1. package com.bigdata;
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;
8. import java.io.IOException;
10. /**
11. *
12. * Mapper中的四个泛型跟什么照应：
13. * 1、LongWritable  行偏移量，一般都是LongWritable ，这一行的数据是从第几个字符开始计算的，因为数据量很多这个值也会很大，所以使用Long
14. * 2、Text     指的是这一行数据
15. * 3、Text     Map任务输出的Key值的类型           单词
16. * 4、IntWritable     Map任务输出的Key值的类型     1
17. *
18. */
19. public class WordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
20.    
21.     // Ctrl + o  可以展示哪些方法可以重写
22.     @Override
23.     protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
24.         // key 值 指的是行偏移量
25.         // value 指的是 这一行数据
26.         //hello bigdata hello 1999 hello beijing hello
27.         String line = value.toString();
28.         // [hello,bigdata,hello,1999,hello,beijing,hello]
29.         String[] arr = line.split("\\s+");
30.         //  hello-> 1,bigdata->1,hello->1,1999->1,hello->1,beijing->1,hello->1
31.         for (String word: arr) {
32.             context.write(new Text(word),new IntWritable(1));
33.         }
34.         // fori 循环
35.         /*for (int i = 0; i < arr.length ; i++) {
36.             context.write(new Text(arr[i]),new IntWritable(1));
37.         }*/
38.         
39.     }
40. }

复制代码

2）编写Reduce代码

1. package com.bigdata;
3. import org.apache.hadoop.io.IntWritable;
4. import org.apache.hadoop.io.Text;
5. import org.apache.hadoop.mapreduce.Reducer;
7. import java.io.IOException;
9. /**
10. *   reduce 是用来合并的
11. *   reduce四个泛型：
12. *   前两个，跟map的输出类型一样
13. *   后面两个泛型：reduce端的输出类型
14. *     hello 5
15. *     world 2
16. *     ...
17. */
18. public class WordCountReducer extends Reducer<Text, IntWritable,Text, IntWritable> {
20.     // reduce 这个方法，有多少个key值，就会调用多少次
21.     @Override
22.     protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
23.         // reduce 拿到的数据是什么样的呢   hello [1,1,1,1,1]   world [1,1]
24.         // 数据不是你想像的这个样子 hello 1   hello 1  hello 1
25.         int count = 0;
26.         // 第一种写法
27.         for (IntWritable num : values) {
28.             int i = num.get();
29.             count = count + i;
30.         }
31.         // 第二种写法
32.         /*Iterator<IntWritable> iterator = values.iterator();
33.         while(iterator.hasNext()){
34.             int i = iterator.next().get();
35.             count = count + i;
36.         }*/
37.         // hello 5
38.         context.write(key,new IntWritable(count));
39.     }
40. }

复制代码

3）编写测试代码

1. package com.bigdata.mr1;
3. import org.apache.hadoop.conf.Configuration;
4. import org.apache.hadoop.fs.Path;
5. import org.apache.hadoop.io.IntWritable;
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;
11. import java.io.IOException;
13. public class WordCountDriver {
15.     public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
16.         Configuration configuration = new Configuration();
17.         // 使用本地的文件系统，而不是hdfs
18.         configuration.set("fs.defaultFS","file:///");
19.         // 使用本地的资源（CPU,内存等）, 也可以使用yarn平台跑任务
20.         configuration.set("mapreduce.framework.name","local");
21.         Job job = Job.getInstance(configuration, "wordCount单词统计");
23.         // 指定 map
24.         job.setMapperClass(WordCountMapper.class);
25.         // hello 1
26.         job.setMapOutputKeyClass(Text.class);
27.         job.setMapOutputValueClass(IntWritable.class);
30.         // 设置reduceTask的数量
31.         // reduce的数量决定了reduceTask的任务数量，每一个任务，结束后都会产生一个文件 part-r-xxxxx
32.         // 结论：reduceTask的数量可以和分区数量不一致，但是没有意义，一般两者保持一致。
33.         job.setNumReduceTasks(1);
36.         // 指定 reduce
37.         job.setReducerClass(WordCountReducer.class);
38.         // hello 5
39.         job.setOutputKeyClass(Text.class);
40.         job.setOutputValueClass(IntWritable.class);
42.         // 此处也可以使用绝对路径
43.         FileInputFormat.setInputPaths(job,"../MapReduceDemo/mr01/input/");
44.         FileOutputFormat.setOutputPath(job,new Path("../MapReduceDemo/mr01/output"));
46.         boolean result = job.waitForCompletion(true);
48.         // 返回结果如果为true表示任务成功了，正常退出，否则非正常退出
49.         System.exit(result?0:-1);
50.     }
51. }

复制代码

4、遇到的错误：

1、输出路径已经存在

2、假如出现如下问题，如何解决？

1. Exception in thread "main" java.lang.UnsatisfiedLinkError: org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Ljava/lang/String;I)Z
2.         at org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Native Method)
3.         at org.apache.hadoop.io.nativeio.NativeIO$Windows.access(NativeIO.java:793)
4.         at org.apache.hadoop.fs.FileUtil.canRead(FileUtil.java:1215)
5.         at org.apache.hadoop.fs.FileUtil.list(FileUtil.java:1420)
6.         at org.apache.hadoop.fs.RawLocalFileSystem.listStatus(RawLocalFileSystem.java:601)
7.         at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:1972)
8.         at org.apache.hadoop.fs.FileSystem.listStatus(FileSystem.java:2014)
9.         at org.apache.hadoop.fs.FileSystem$4.<init>(FileSystem.java:2180)
10.         at org.apache.hadoop.fs.FileSystem.listLocatedStatus(FileSystem.java:2179)
11.         at org.apache.hadoop.fs.ChecksumFileSystem.listLocatedStatus(ChecksumFileSystem.java:783)
12.         at org.apache.hadoop.mapreduce.lib.input.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:320)
13.         at org.apache.hadoop.mapreduce.lib.input.FileInputFormat.listStatus(FileInputFormat.java:279)
14.         at org.apache.hadoop.mapreduce.lib.input.FileInputFormat.getSplits(FileInputFormat.java:404)
15.         at org.apache.hadoop.mapreduce.JobSubmitter.writeNewSplits(JobSubmitter.java:310)
16.         at org.apache.hadoop.mapreduce.JobSubmitter.writeSplits(JobSubmitter.java:327)
17.         at org.apache.hadoop.mapreduce.JobSubmitter.submitJobInternal(JobSubmitter.java:200)
18.         at org.apache.hadoop.mapreduce.Job$11.run(Job.java:1571)
19.         at org.apache.hadoop.mapreduce.Job$11.run(Job.java:1568)
20.         at java.security.AccessController.doPrivileged(Native Method)
21.         at javax.security.auth.Subject.doAs(Subject.java:422)
22.         at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1878)
23.         at org.apache.hadoop.mapreduce.Job.submit(Job.java:1568)
24.         at org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:1589)
25.         at mr02.WordCountDriver.main(WordCountDriver.java:58)

复制代码

假如之前在windows中的hadoop中拷贝过hadoop.dll等文件，是不会出现这个问题的。必要进行安装包的补丁配置。
解决方案：

1. 将之前的 hadoop.dll 文件拷贝在  C:/Windows/System32文件夹下一份，可以不重启试试，假如不管用，重启一下。

复制代码

三、MR 中的自界说分区

需求：不仅单词统计，还必要将a-p 的单词存放在一起，q-z的单词存放在一起，其他单词存放在另一个文件中。

假如要完成以上的需求：就必要引入新的组件Partitioner。
1、编写代码

1. package com.bigdata;
3. import org.apache.hadoop.io.IntWritable;
4. import org.apache.hadoop.io.Text;
5. import org.apache.hadoop.mapreduce.Partitioner;
7. /**
8. *
9. * Map任务 --> Partitioner  --> Reducer
10. *   Partitioner 其实就是Map端的输出
11. */
12. public class WordCountPartitioner extends Partitioner<Text, IntWritable> {
14.     // 分区的区号，一定是从0开始的，中间不能断   0 1 2 3 4..
15.     @Override
16.     public int getPartition(Text text, IntWritable intWritable, int i) {
17.         // text就是一个单词
18.         String letter = text.toString();
19.         char c = letter.charAt(0);
20.         if(c>='a' && c <='p'){
21.             return 0;
22.         }else if(c>='q' && c <='z'){
23.             return 1;
24.         }else{
25.             return 2;
26.         }
28.     }
29. }

复制代码

2、使用分区代码

1. package com.bigdata;
3. import org.apache.hadoop.conf.Configuration;
4. import org.apache.hadoop.fs.Path;
5. import org.apache.hadoop.io.IntWritable;
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;
11. /**
12. * @Author laoyan
13. * @Description TODO
14. * @Date 2022/8/1 10:02
15. * @Version 1.0
16. */
17. public class WordCountDriver2 {
18.     public static void main(String[] args) throws Exception {
19.         
20.         Configuration configuration = new Configuration();
21.         // 使用本地的文件系统，而不是hdfs
22.         configuration.set("fs.defaultFS","file:///");
23.         // 使用本地的资源（CPU,内存等）, 也可以使用yarn平台跑任务
24.         configuration.set("mapreduce.framework.name","local");
25.         Job job = Job.getInstance(configuration, "单词统计WordCount");
26.         // map任务的设置
27.         job.setMapperClass(WordCountMapper.class);
28.         job.setMapOutputKeyClass(Text.class);
29.         job.setMapOutputValueClass(IntWritable.class);
30.         
31.         // 指定分区的类是哪一个
32.         job.setPartitionerClass(WordCountPartitioner.class);
33.         // 还要执行 reduce的数量  因为一个reduce 就会产生一个结果文件
34.         job.setNumReduceTasks(3);
35.         
36.         // reduce任务的设置
37.         job.setReducerClass(WordCountReducer.class);
38.         job.setOutputKeyClass(Text.class);
39.         job.setOutputValueClass(IntWritable.class);
40.         
41.         // 设置要统计的数据的路径，结果输出路径
42.         FileInputFormat.setInputPaths(job,new Path("mr01/input"));
43.         // ouput文件夹一定不要出现，否则会报错
44.         FileOutputFormat.setOutputPath(job,new Path("mr01/output2"));
45.         // 等待任务执行结束
46.         boolean b = job.waitForCompletion(true);
47.         // 此处是一定要退出虚拟机的
48.         System.exit(b ? 0:-1);
49.     }
50. }

复制代码

假如分区文件，指定的是3个分区，reduce的使命设置的是1个，结果就是产生1个输出文件，三个分区的结果在一个文件中。
假如是3个分区，reduce的使命设置的是5个，输出结果就是5个文件，其中三个是有值的，背面两个是没有值的空文件。

综上所述：分区的数量，肯定要跟reduce的使命数量一致，服从是最高的。

补充：如何自动删除输出文件夹

1. // 设置输入路径
2. FileInputFormat.setInputPaths(job,"./data/");
3. // 设置输出路径
4. // 如何自动删除文件夹
5. // 判断该文件夹是否存在，如何存在，删除。
6. String outPath = "./data/output03";
7. Path path = new Path(outPath);
8. FileSystem fileSystem = FileSystem.get(conf);
9. if(fileSystem.exists(path)){
10.     System.out.println("文件夹已存在");
11.     fileSystem.delete(path,true);
12. }

复制代码

再将从前的代码简化一下写法：

1. package com.bigdata;
3. import org.apache.hadoop.conf.Configuration;
4. import org.apache.hadoop.fs.FileSystem;
5. import org.apache.hadoop.fs.Path;
6. import org.apache.hadoop.io.IntWritable;
7. import org.apache.hadoop.io.LongWritable;
8. import org.apache.hadoop.io.Text;
9. import org.apache.hadoop.mapreduce.Job;
10. import org.apache.hadoop.mapreduce.Mapper;
11. import org.apache.hadoop.mapreduce.Reducer;
12. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
13. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
15. import java.io.IOException;
17. //  java 1
19. /**
20. *   四个泛型分别是什么意思？
21. *   第1和第2 个泛型指的是输入类型，固定的LongWritable,Text
22. *
23. *   第一个泛型指的是  文件的开始位置，就是偏移量  我们一般不用
24. *   第二个泛型就是： 一行数据
25. *
26. *   也就意味着：一行数据执行一个map 方法
27. *   第3和4 的泛型
28. *   是输出类型
29. *   第三个泛型：输出key的类型  String --> Text
30. *   第四个泛型：输出value的类型  int  --> IntWritable
31. *
32. */
33. class WordCountMapper2 extends Mapper<LongWritable,Text, Text, IntWritable> {
35.     // ctrl + o  提示哪些方法可以重写
36.     // hello bigdata hello 1999 hello beijing hello
37.     // hello 1
38.     // bigdata 1
39.     // hello 1  ....
40.     @Override
41.     protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
42.         System.out.println("本次的偏移量是："+key.get());
43.         // 如何获取一行数据
44.         String line = value.toString();
45.         String[] arr = line.split("\\s+");
46.         for (String word : arr) {
47.             // 循环一个个的单词，并且将单词往外写出   word 1  bigdata 1.....
48.             context.write(new Text(word),new IntWritable(1));
49.         }
52.     }
53. }
55. class WordCountReducer2 extends Reducer<Text, IntWritable,Text, IntWritable> {
57.     // reduce 是聚合的意思
58.     // key 指的是一个单词
59.     // values [1,1,1,1,1,1]
60.     // hello  [1,1,1,1,1,1,1,1,1]
61.     @Override
62.     protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
63.         int sum = 0;
64.         for (IntWritable value : values) {
65.             sum += value.get();
66.         }
67.         context.write(key,new IntWritable(sum));
68.     }
69. }
72. public class WcDriver {
74.     public static void main(String[] args) throws Exception{
75.         // 以下都是固定写法
76.         Configuration conf = new Configuration();
78.         // 使用本地的文件系统，而不是hdfs
79.         conf.set("fs.defaultFS","file:///");
80.         // 使用本地的资源（CPU,内存等）, 也可以使用yarn平台跑任务
81.         conf.set("mapreduce.framework.name","local");
83.         Job job = Job.getInstance(conf, "单词统计");
84.         job.setMapperClass(WordCountMapper2.class);
88.         job.setReducerClass(WordCountReducer2.class);
89.         // bigdata 1  map类的输出类型
90.         job.setMapOutputKeyClass(Text.class);
91.         job.setMapOutputValueClass(IntWritable.class);
92.         // reduce类的输出类型  bigdata 10
93.         job.setOutputKeyClass(Text.class);
94.         job.setOutputValueClass(IntWritable.class);
96.         FileInputFormat.setInputPaths(job,new Path("./mr01/input"));
98.         // 每次运行的时候，假如输出文件夹存在，就会报错，严重影响体验
99.         // 是否可以判断一下，假如存在，直接删除
100.         Path path = new Path("./mr01/output");
102.         FileSystem fileSystem = FileSystem.get(conf);
103.         if(fileSystem.exists(path)){
104.             fileSystem.delete(path,true);
105.         }
106.         FileOutputFormat.setOutputPath(job,path);
108.         boolean b = job.waitForCompletion(true);
109.         System.exit(b?0:-1);
110.     }
111. }

复制代码

四、序列化

啥是序列化？
jvm中的一个对象，不是类，假如你想把一个对象，生存到磁盘上，必须序列化，你把文件中的对象进行恢复，是不是的反序列化。
假如你想把对象发送给另一个服务器，必要通过网络传输，也必须序列化，到另一侧要反序列化。
大数据中，必要用到序列化，因为数据在传输过程中，必要用到。

1. 说到序列化，我们想到了Java的序列化。一个类实现了Serializable 接口即可。
3. Java对象什么时候需要序列化？
4. 1）需要保存到本地的时候
5. 2）需要在网络之间传输的时候
7. package com.bigdata;
9. import java.io.Serializable;
10. /**
11. * @Author laoyan
12. * @Description TODO
13. * @Date 2022/8/1 11:43
14. * @Version 1.0
15. */
16. public class User implements Serializable {
18.     private String name;
19.     private int age;
21.     public String getName() {
22.         return name;
23.     }
25.     public void setName(String name) {
26.         this.name = name;
27.     }
29.     public int getAge() {
30.         return age;
31.     }
33.     public void setAge(int age) {
34.         this.age = age;
35.     }
36. }

复制代码

大数据技能Hadoop并没有采用java的序列化机制，而是自己又整了一套自己的序列化机制。为什么？

1. Java的序列化携带的信息太多了，文件太大了，不便于在网络之间传输。
2. User  使用Java  --> 100KB
3. User  使用大数据的序列化 --> 5KB
5. 大数据采用的序列化机制是  Writable 接口。
7. 为什么非得序列化呢？因为需要在网路之间传输。

复制代码

Java数据范例	Hadoop序列化的数据范例	释义
byte	ByteWritable	字节范例
short	ShortWritable	短整型
int	IntWritable	整型
long	LongWritable	长整型
float	FloatWritable	单精度浮点型
double	DoubleWritable	双精度浮点型
boolean	BooleanWritable	布尔型
String	Text	字符串
array	ArrayWritable	数组
Map	MapWritable	Map
null	NullWritable	空

java的八大根本数据范例： byte short int long float double char boolean
只必要记住：String --> Text即可。null --> NullWritable，仅仅是为了在某个地方占位，符合语法而已。

1、测试java序列化 VS Hadoop序列化大小比力

1. package com.bigdata;
3. import java.io.Serializable;
5. /**
6. * @Author laoyan
7. * @Description TODO
8. * @Date 2022/8/1 11:43
9. * @Version 1.0
10. */
11. public class User implements Serializable {
13.     private String name;
14.     private int age;
16.     public User(String name, int age) {
17.         this.name = name;
18.         this.age = age;
19.     }
21.     public String getName() {
22.         return name;
23.     }
25.     public void setName(String name) {
26.         this.name = name;
27.     }
29.     public int getAge() {
30.         return age;
31.     }
33.     public void setAge(int age) {
34.         this.age = age;
35.     }
36. }

复制代码

1. package com.bigdata;
3. import org.apache.hadoop.io.Writable;
5. import java.io.DataInput;
6. import java.io.DataOutput;
7. import java.io.IOException;
9. /**
10. * @Author laoyan
11. * @Description TODO
12. * @Date 2022/8/1 14:06
13. * @Version 1.0
14. */
15. public class UserWritable implements Writable {
17.     private String name;
18.     private int age;
20.     public UserWritable(String name, int age) {
21.         this.name = name;
22.         this.age = age;
23.     }
25.     // 序列化
26.     @Override
27.     public void write(DataOutput out) throws IOException {
28.         out.writeUTF(name);
29.         out.writeInt(age);
30.     }
32.     // 反序列化
33.     @Override
34.     public void readFields(DataInput in) throws IOException {
36.         // 进行反序列化的时候，读取的顺序一定要跟序列化的时候的顺序一致，否则报错
37.         name = in.readUTF();
38.         age = in.readInt();
39.     }
41.     public String getName() {
42.         return name;
43.     }
45.     public void setName(String name) {
46.         this.name = name;
47.     }
49.     public int getAge() {
50.         return age;
51.     }
53.     public void setAge(int age) {
54.         this.age = age;
55.     }
56. }

复制代码

1. package com.bigdata;
3. import java.io.FileNotFoundException;
4. import java.io.FileOutputStream;
5. import java.io.ObjectOutputStream;
7. /**
8. * @Author laoyan
9. * @Description TODO
10. * @Date 2022/8/1 14:11
11. * @Version 1.0
12. */
13. public class TestXLH {
15.     public static void main(String[] args) throws Exception {
16.         User user = new User("zhangsan",20);
17.         ObjectOutputStream objectOutputStream = new ObjectOutputStream(new FileOutputStream("D:/user1.txt"));
18.         objectOutputStream.writeObject(user);
19.         objectOutputStream.close();
21.         UserWritable user2 = new UserWritable("zhangsan",20);
22.         ObjectOutputStream objectOutputStream2 = new ObjectOutputStream(new FileOutputStream("D:/user2.txt"));
23.         // 此时是序列化对象去write 对象流，此处需要注意
24.         user2.write(objectOutputStream2);
25.         objectOutputStream2.close();
27.     }
28. }

复制代码

java序列化的结果：

hadoop序列化的结果：

2、自界说序列化案例--手机流量统计

需求：假定拿到了一些关于手机流量的日志文件，统计每个手机号码的上行流量，下行流量，以及总流量。

读懂日志文件：

1. 1363157995033         15920133257        5C-0E-8B-C7-BA-20:CMCC        120.197.40.4        sug.so.360.cn        信息安全        20        20        3156        2936        200
2.    
3. 第二列是手机号码       15920133257
4. 倒数第三列是上行流量   3156
5. 倒数第二列是下行流量   2936

复制代码

思绪：

1. map任务：
2.    15989002119   1232 3456
3.    15989002119   2343 34343
4.     ......
5.    13726238888   3243  23432
6.    13726238888  4343 4343
7. reduce 任务：
8. 15989002119   对相同的手机号码进行合并
9.   Key:  15989002119    values: [{15989002119,34343,34343},{15989002119,8989,34343}....]
10.   
11. 此处：手机号码，上行流量，下行流量，总流量
12. 此时就需要自定义数据类型了。

复制代码

统计的结果：

代码演示：

1. package com.bigdata.phoneflow;
3. import org.apache.hadoop.io.Writable;
5. import java.io.DataInput;
6. import java.io.DataOutput;
7. import java.io.IOException;
9. /**
10. * @Author laoyan
11. * @Description TODO
12. * @Date 2022/8/1 14:29
13. * @Version 1.0
14. */
15. // 自定义一个数据类型
16. public class PhoneFlowWritable implements Writable {
18.     private String phone;
19.     private int upFlow;
20.     private int downFlow;
22.     // 此处需要指定一个空参数的构造方法，否则报错：
23.     // java.lang.NoSuchMethodException: com.bigdata.phoneflow.PhoneFlowWritable.<init>()
24.     public PhoneFlowWritable() {
25.     }
27.     public PhoneFlowWritable(String phone, int upFlow, int downFlow) {
28.         this.phone = phone;
29.         this.upFlow = upFlow;
30.         this.downFlow = downFlow;
31.     }
33.     public String getPhone() {
34.         return phone;
35.     }
37.     public void setPhone(String phone) {
38.         this.phone = phone;
39.     }
41.     public int getUpFlow() {
42.         return upFlow;
43.     }
45.     public void setUpFlow(int upFlow) {
46.         this.upFlow = upFlow;
47.     }
49.     public int getDownFlow() {
50.         return downFlow;
51.     }
53.     public void setDownFlow(int downFlow) {
54.         this.downFlow = downFlow;
55.     }
57.     @Override
58.     public void write(DataOutput out) throws IOException {
60.         out.writeUTF(phone);
61.         out.writeInt(upFlow);
62.         out.writeInt(downFlow);
63.     }
65.     @Override
66.     public void readFields(DataInput in) throws IOException {
68.         phone =  in.readUTF();
69.         upFlow = in.readInt();
70.         downFlow = in.readInt();
71.     }
72. }

复制代码

1. package com.bigdata.phoneflow;
3. import com.bigdata.WordCountMapper;
4. import com.bigdata.WordCountPartitioner;
5. import com.bigdata.WordCountReducer;
6. import org.apache.hadoop.conf.Configuration;
7. import org.apache.hadoop.fs.Path;
8. import org.apache.hadoop.io.IntWritable;
9. import org.apache.hadoop.io.LongWritable;
10. import org.apache.hadoop.io.Text;
11. import org.apache.hadoop.mapreduce.Job;
12. import org.apache.hadoop.mapreduce.Mapper;
13. import org.apache.hadoop.mapreduce.Reducer;
14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
15. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
17. import java.io.IOException;
18. import java.util.Iterator;
20. /**
21. * @Author laoyan
22. * @Description TODO
23. * @Date 2022/8/1 14:33
24. * @Version 1.0
25. */
26. class PhoneFlowMapper extends Mapper<LongWritable, Text,Text,PhoneFlowWritable> {
28.     // 将每一句话，都变为   手机号码 -->  PhoneFlowWritable对象
29.     @Override
30.     protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, PhoneFlowWritable>.Context context) throws IOException, InterruptedException {
32.         String line = value.toString();
33.         String[] arr = line.split("\\s+");
34.         if(arr.length == 11){
35.             String phone = arr[1];
36.             // 正则表达式，判断手机号
37.             if (phone.matches("^1[3456789][0-9]{9}$")){
38.                 int upFlow = Integer.parseInt(arr[arr.length-3]);
39.             int downFlow = Integer.parseInt(arr[arr.length-2]);
40.             System.out.println(phone+","+upFlow+","+downFlow);
41.             context.write(new Text(phone),new PhoneFlowWritable(phone,upFlow,downFlow));
42.             }
44.         }
45.     }
46. }
47. //   手机号 --> 流量数据PhoneFlowWritable      手机号码 --> 统计的结果
48. class PhoneFlowReducer extends Reducer<Text,PhoneFlowWritable,Text,Text> {
50.     @Override
51.     protected void reduce(Text key, Iterable<PhoneFlowWritable> values, Reducer<Text, PhoneFlowWritable, Text, Text>.Context context) throws IOException, InterruptedException {
53.         int upFlowNum = 0;
54.         int downFlowNum = 0;
55.         Iterator<PhoneFlowWritable> iterator = values.iterator();
56.         while(iterator.hasNext()){
57.             PhoneFlowWritable phoneFlowWritable = iterator.next();
58.             upFlowNum += phoneFlowWritable.getUpFlow();
59.             downFlowNum += phoneFlowWritable.getDownFlow();
60.         }
61.         StringBuffer sb = new StringBuffer();
62.         sb.append("手机号"+key+"流量统计：");
63.         sb.append("上行流量是:"+upFlowNum);
64.         sb.append("下行流量是:"+downFlowNum);
65.         sb.append("总的流量是:"+(upFlowNum + downFlowNum));
66.         context.write(key,new Text(sb.toString()));
68.     }
69. }
73. public class PhoneFlowDriver {
75.     public static void main(String[] args) throws Exception{
77.         Configuration configuration = new Configuration();
78.         // 使用本地的文件系统，而不是hdfs
79.         configuration.set("fs.defaultFS","file:///");
80.         // 使用本地的资源（CPU,内存等）, 也可以使用yarn平台跑任务
81.         configuration.set("mapreduce.framework.name","local");
82.         Job job = Job.getInstance(configuration, "手机流量统计");
83.         // map任务的设置
84.         job.setMapperClass(PhoneFlowMapper.class);
85.         job.setMapOutputKeyClass(Text.class);
86.         job.setMapOutputValueClass(PhoneFlowWritable.class);
88.         // reduce任务的设置
89.         job.setReducerClass(PhoneFlowReducer.class);
90.         job.setOutputKeyClass(Text.class);
91.         job.setOutputValueClass(Text.class);
93.         // 设置要统计的数据的路径，结果输出路径
94.         FileInputFormat.setInputPaths(job,new Path("mr01/phoneFlow/input"));
95.         // ouput文件夹一定不要出现，否则会报错
96.         FileOutputFormat.setOutputPath(job,new Path("mr01/phoneFlow/output"));
97.         // 等待任务执行结束
98.         boolean b = job.waitForCompletion(true);
99.         // 此处是一定要退出虚拟机的
100.         System.exit(b ? 0:-1);
101.     }
102. }

复制代码

注意这个案例中有一个关于迭代器的用法，迭代器实在也可以通过foreach循环进行遍历：

1. for (PhoneFlowWritable phoneFlow : values) {
2.         sumUpFlow += phoneFlow.getUpFlow();
3.         sumDownFlow += phoneFlow.getDownFlow();
4. }

复制代码

随堂代码如下：

1. package com.bigdata.phoneflow;
3. import org.apache.hadoop.conf.Configuration;
4. import org.apache.hadoop.fs.FileSystem;
5. import org.apache.hadoop.fs.Path;
6. import org.apache.hadoop.io.IntWritable;
7. import org.apache.hadoop.io.LongWritable;
8. import org.apache.hadoop.io.NullWritable;
9. import org.apache.hadoop.io.Text;
10. import org.apache.hadoop.mapreduce.Job;
11. import org.apache.hadoop.mapreduce.Mapper;
12. import org.apache.hadoop.mapreduce.Reducer;
13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
16. import java.io.IOException;
18. /**
19. * @基本功能:
20. * @program:MapReduce
21. * @author: 闫哥
22. * @create:2024-01-22 15:02:25
23. **/
24. // 一个java 文件中可以有多个class ，但是只能有一个public class ,而且这个类名必须和java 的文件名一样
26.     //  手机号  一个自定义的类
27. class PhoneFlowMapper extends Mapper<LongWritable, Text,Text,PhoneFlowWritable> {
29.     // ctrl + o
31.     @Override
32.     protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, PhoneFlowWritable>.Context context) throws IOException, InterruptedException {
33.         String line = value.toString();
34.         String[] arr = line.split("\\s+");
35.         if(arr.length == 11){
36.             String phone = arr[1];
37.             // 正则表达式，判断手机号
38.             if (phone.matches("^1[3456789][0-9]{9}$")){
39.                 int upFlow = Integer.parseInt(arr[arr.length-3]);
40.             int downFlow = Integer.parseInt(arr[arr.length-2]);
41.                 context.write(new Text(phone),new PhoneFlowWritable(phone,upFlow,downFlow));
42.             }
44.         }
46.     }
47. }
48. class PhoneFlowReducer extends Reducer<Text,PhoneFlowWritable, NullWritable,Text>{
49.     @Override
50.     protected void reduce(Text phone, Iterable<PhoneFlowWritable> values, Reducer<Text, PhoneFlowWritable, NullWritable, Text>.Context context) throws IOException, InterruptedException {
51.         long sumUpFlow = 0;
52.         long sumDownFlow = 0;
53.         for (PhoneFlowWritable phoneFlow : values) {
54.             sumUpFlow += phoneFlow.getUpFlow();
55.             sumDownFlow += phoneFlow.getDownFlow();
56.         }
57.         String str = "手机号为："+phone+",上行总流量："+sumUpFlow+",下行总流量："+sumDownFlow+",总量："+(sumUpFlow+sumDownFlow);
58.         context.write(NullWritable.get(),new Text(str));
59.     }
60. }
61. public class PhoneFlowDriver {
63.     public static void main(String[] args) throws Exception{
65.         // 先获取一个Job对象
66.         Configuration conf = new Configuration();
67.         // 第一个：使用本地文件系统
68.         conf.set("fs.defaultFS","file:///");
69.         // 第二个：使用本地资源
70.         conf.set("mapreduce.framework.name","local");
71.         Job job = Job.getInstance(conf, "流量统计");
73.         // 设置 map
74.         job.setMapperClass(com.bigdata.phoneflow.PhoneFlowMapper.class);
75.         job.setMapOutputKeyClass(Text.class);
76.         job.setMapOutputValueClass(PhoneFlowWritable.class);
80.         // 设置reduce
81.         job.setReducerClass(com.bigdata.phoneflow.PhoneFlowReducer.class);
82.         job.setOutputKeyClass(NullWritable.class);
83.         job.setOutputValueClass(Text.class);
85.         // 设置输入路径
86.         FileInputFormat.setInputPaths(job,"./data/phoneflow/");
87.         // 设置输出路径
88.         // 如何自动删除文件夹
89.         // 判断该文件夹是否存在，如何存在，删除。
90.         String outPath = "./data/phoneflowresult/";
91.         Path path = new Path(outPath);
92.         FileSystem fileSystem = FileSystem.get(conf);
93.         if(fileSystem.exists(path)){
94.             System.out.println("文件夹已存在");
95.             fileSystem.delete(path,true);
96.         }
98.         FileOutputFormat.setOutputPath(job,path);
100.         boolean result = job.waitForCompletion(true);
101.         System.exit(result?0:-1);
102.     }
103. }

复制代码

3、假如我们在编写代码的时候不出结果，也不报错，怎么办？

1. # Global logging configuration
2. #  Debug   info   warn  error
3. log4j.rootLogger=WARN, stdout
4. # MyBatis logging configuration...
5. log4j.logger.org.mybatis.example.BlogMapper=TRACE
6. # Console output...
7. log4j.appender.stdout=org.apache.log4j.ConsoleAppender
8. log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
9. log4j.appender.stdout.layout.ConversionPattern=%5p [%t] - %m%n

复制代码

五、关于片和块

1. 假如我现在500M这样的数据，如何存储？
2. 500M = 128M + 128M + 128M + 116M  分为四个块进行存储。
3. 计算的时候，是按照片儿计算的，而不是块儿。
4. 块是物理概念，一个块就是128M ,妥妥的，毋庸置疑。
5. 片是逻辑概念，一个片大约等于一个块。
7. 假如我现在需要计算一个300M的文件，这个时候启动多少个MapTask任务？答案是有多少个片儿，就启动多少个任务。
8. 一个片儿约等于 一个块，但是最大可以 128M*1.1倍= 140.8
9. 300M    128+128+44
10. 128M  启动一个Map任务进行读取
11. 172M  172M  和   128M * 1.1 =140.8M 进行比较，如果大于 ,继续进行切割
12. 128M  启动一个任务Map任务
13. 剩余44M   剩余的44M 和  128M*1.1倍比较，小于这个值，剩余的44M 就单独起一个Map任务
14. 300m的数据，分给了3个MapTask任务进行处理。
16. 如果是260M的数据，由多少个Map任务处理？
17. 128M   第一个任务
18. 132M   跟 128M * 1.1 进行比较，发现小于这个值，直接一个Map任务搞定，不在启动第三个任务了。

复制代码

比如班里的同砚一起搬砖，每人规定搬3块，假定砖还剩4块，到某个同砚了，他就直接搬完即可，没须要让另一个同砚因为一块砖，而专门跑一趟。

1. 1、什么是片，什么是块？
2. 块是物理概念，片是逻辑概念。一般片 = 块的，但是到最后一次的时候，有可能片> 块，但是绝对不能超过块的1.1倍。
3. 2、mapreduce 启动多少个MapTask任务？
4. 跟片有关系，有多少个片，就启动多少个map任务。跟块儿无关。

复制代码

六、Shuffle 过程【必问】

1. MapReduce的Shuffle过程指的是MapTask的后半程，以及ReduceTask的前半程，共同组成的。
2. 从MapTask中的map方法结束，到ReduceTask中的reduce方法开始，这个中间的部分就是Shuffle。是MapReduce的核心，心脏。

复制代码

map端：

1. 1、map中的context.write方法，对外写出的时候，其实是写入到了一个环形缓冲区内（内存形式的），这个环形缓冲区大小是100M,可以通过参数设置。如果里面的数据大于80M,就开始溢写（从内存中将数据写入到磁盘上）。溢写的文件存放地址可以设置。
2. 2、在溢写过程中，环形缓冲区不会停止工作，是会利用剩余的20%继续存入环形缓冲区的。除非是环形缓冲区的内存满了，map任务就被阻塞了。
3. 在溢写出来的文件中，是排过序的，排序规则：快速排序算法。在排序之前，会根据分区的算法，对数据进行分区。是在内存中，先分区，在每一个分区中再排序，接着溢写到磁盘上的。
4. 3、溢写出来的小文件需要合并为一个大文件，因为每一个MapTask只能有一份数据。就将相同的分区文件合并，并且排序（此处是归并排序）。每次合并的时候是10个小文件合并为一个大文件，进行多次合并，最终每一个分区的文件只能有一份。
5. 假如100个小文件，需要合并几次呢？
6. 100  每10分合并一次，第一轮：100个文件合并为了10个文件，这10个文件又合并为一个大文件，总共合并了11次。
8. 4、将内存中的数据，溢写到磁盘上，还可以指定是否需要压缩，以及压缩的算法是什么。

复制代码

reduce端：

1. 1、reduce端根据不同的分区，拉取每个服务器上的相同的分区的数据。
2. reduce任务有少量复制线程，因此能够并行取得map输出。默认值是5个线程，但这个默认值可以修改设置mapreduce.reduce.shuffle. parallelcopies 属性即可。
3. 2、如果map上的数据非常的小，该数据会拉取到reduce端的内存中，如果数据量比较大，直接拉取到reduce端的硬盘上。

复制代码

七、Combiner 【可有可无】

这个Combiner是一个优化的代码，对于我们最终的结果没有任何的影响。
map端产生的数据，会被拉去到reduce端进行合并，有可能map端产生的数据非常的大，未便于在网络间传输，那么有没有办法可以缩小map端的数据呢？
之前： java 1 java 1 java 1 传递给reduce
如今： java 3 传递给reduce
Combiner实在就是运行在mapTask中的reducer。 Reducer实在就是合并代码的。Combiner是作用在Map端的。
这个结果不是最终的结果，而是一个暂时的小统计。 最终reduce是会将所有的map结果再次进行汇总才是我们最终想要的统计结果。

1. Combiner 只能用于对统计结果没有影响的场景下。
2. 一般只用于  统计之和，统计最大值最小值的场景下。统计平均值等情况是不能用的。

复制代码

在代码中如何使用？

Combiner起作用的地方：

Combiner 实在作用于两个地方，一个是环形缓冲区溢写磁盘的时候，除了分区，排序之外，还可以做合并操作，将内存中的 hello 1 hello 1 hello 1 会合并为 hello 3
第二个位置是小文件合并为MapTask的大文件的时候，会将多个 hello 的值相加 hello 19，但是这个不是最终的答案，最终答案是将多个MapTask使掷中的hello 进行合并才是最终的结果。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。