Hadoop

HDFS

配置文件

core-site.xml

	 
	 
		 fs.defaultFS
		 hdfs://hadoop102:8020
	 
	 
	 
		 hadoop.tmp.dir
		 /opt/module/hadoop-3.1.3/data
	 
	 
	 
		 hadoop.http.staticuser.user
		 atguigu
	 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

hdfs-site.xml

	


	
	
		 dfs.namenode.http-address
		 hadoop102:9870
	 
	
	 
		 dfs.namenode.secondary.http-address
		 hadoop104:9868
	 

1
2
3
4
5
6
7
8
9
10
11
12
13
14

yarn-site.xml

	 
	 
		 yarn.nodemanager.aux-services
		 mapreduce_shuffle
	 
	 
	 
		 yarn.resourcemanager.hostname
		 hadoop103
	 
	 
	 
	 yarn.nodemanager.env-whitelist
	 
	JAVA_HOME,HADOOP_COMMON_HOME,HADOOP_HDFS_HOME,HADOOP_CO
	NF_DIR,CLASSPATH_PREPEND_DISTCACHE,HADOOP_YARN_HOME,HADOOP_MAP
	RED_HOME
	 


	
	
	 yarn.log-aggregation-enable
	 true
	
	
	 
	 yarn.log.server.url 
	 http://hadoop102:19888/jobhistory/logs
	
	
	
	 yarn.log-aggregation.retain-seconds
	 604800
	


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

mapred-site.xml

 
 mapreduce.framework.name
 yarn
 
 

 mapreduce.jobhistory.address
 hadoop102:10020



 mapreduce.jobhistory.webapp.address
 hadoop102:19888


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

产生背景

一个操作系统存不下所有数据
1

定义

HDFS（Hadoop Distributed File System）他是一个分布式文件系统
1

优缺点

优点

1. 高容错性
   数据自动保存多个副本，他通过增加副本的形式，提高容错性
   某个副本丢失后他可以自动恢复
2. 适合处理大数据
   数据规模：GB、TB、PB
   文件规模：百万规模以上的文件数量
3. 构建机器廉价

缺点

1. 不适合低延时数据访问，比如毫秒级的存储数据

2. 无法高效对大量小文件进行存储
   存储大量小文件的话，他会占用NameNode 大量内存来存储文件目录和块信息。这样是不可取，因为NameNode的内存是有限的

   小文件存储的寻址时间会超过读取时间，他违反了HDFS的设计目标

3. 不支持并发写入，文件随机修改
   一个文件只能有一个写，不允许多线程同时写
   仅支持数据append，不支持文件的随机修改

HDFS组成架构

1. NameNode：就是Master，他是一个主管，
   管理HDFS的名称空间
   配置副本策略
   管理数据块映射信息
   处理客户端读写请求
2. DataNode：执行者
   存储实际数据块
   执行数据块的读写操作
3. Client：客户端
   文件切分。Client将文件切成一个一个Block块进行上传
   与NameNode交互，获取文件位置信息
   与DataNode交互，读取或写入数据
   提供命令管理和访问HDFS
4. Secondary NameNode：并非NameNode热备（需要手动设置）。
   辅助NameNode，分档工作量，比如定期合并Fsimage和Edits，并推送NameNode
   在紧急情况下，可辅助NameNode

HDFS文件块大小

HDFS的文件在物理上是分块存储的（Block），
块的大小可以通过配置参数（dfs.blocksize）来规定，默认大小在hadoop3版本是128M
 为什么块的大小不能设置太小和过大？
 （1）HDFS的块设置太小，增加寻址时间
 （2）太大，从磁盘传输数据的时间会明显大于定位这个块开始位置所需要的时间。处理速度慢
 总结HDFS块的大小设置主要取决于磁盘传输速率
1
2
3
4
5
6

HDFS读写流程

1. 客户端通过Distributed FileSystem模块想NameNode请求上传文件，NameNode价差目标文件是否存在，父目录是否存在
2. NameNode返回是否可以上传
3. 客户端请求第一个Block上传到那几个DataNode服务器上
4. NameNode返回3个DataNode，分别dn1，dn2，dn3
5. 客户端通过FSDataOutputStream模块请求dn1上传数据，dn1收到请求会继续调用dn2，然后dn2调用dn3，将这个通信管道建立完成
6. dn1，dn2，dn3逐级应答客户端
7. 客户端开始往dn1上传第一个Block（先从磁盘读取数据放到一个本地内存缓存），以Packet为单位，dn1收到一个Packet就会传给dn2，dn2传递给dn3，dn1每传一个packet会放入一个应答队列等待应答
8. 当一个Block传输完成之后，客户端再次请求NameNode上传第二个Block服务器（重复执行3-7）

网络拓扑-节点距离计算

在HDFS写数据的过程中，NameNode会选择距离带上传数据最近距离的DataNode接收数据。如何计算？
节点距离：两个节点到达最近共同祖先的距离总和

机架感知

第一个副本在Client所处的节点上，如果客户端在集群外，随机选一个
第二个副本在另一台机架的随机一个节点上
第三个副本在第二个副本所在机架的随机节点
1
2
3

读取数据流程

1. 客户端通过DistributeFileSystem向NameNode请求下载文件，NameNode通过查询数据，找到文件块所在的DataNode地址
2. 挑选一台DataNode（就近原则，然后随机）服务器，请求读取数据
3. DataNode开始传输数据给客户端（从磁盘里面读取数据输入流，以Packet为单位）
4. 客户端以Packet为单位接收，现在本地缓存，然后写入目标文件

NN和2NN工作机制

思考：NameNode中的元数据存储在哪里？
内存中一份数据
FSImage和Edits 两者合并等于内存

SeondaryNamenode专门用于FsImage和Edits的合并
1

第一个阶段：NameNode启动

1. 第一次启动NameNode格式化后，创建Fsimage和Edits文件，如果不是第一次启动，直接加载编辑日志和镜像文件到内存

2. 客户端对元数据进行增删改的请求

3. NameNode记录操作日志，更新滚动日志

4. NameNode在内存中对元数据进行增删改

第二阶段：Secondary NameNode工作

1. Secondary NameNode询问NameNode是否需要CheckPoint。直接带回NameNode检查结果
2. Secondary NameNode请求执行CheckPoint
3. NameNode滚动正在写的Edits日志
4. 将滚动前的编辑日志和镜像文件拷贝到Secondary NameNode
5. Secondary NameNode加载编辑日志和镜像文件到内存，并合并
6. 生成新的镜像文件fsimage.chkpoint
7. 拷贝fsimage.chkpoint到NameNode
8. NameNode将faimage.chkpoint重新命名成fsimage

Fsimage和Edits解析

NameNode被格式化之后 ，将在/opt/moudle/hadoop-3.1.3/data/tmp/dfs/name/current目录中产生如下文件

1. Fsimage文件：HDFS文件系统元数据的一个永久性的检查点，其中包含HDFS文件系统的素有目录和文件inode的序列化信息
2. Edits文件：存放HDFS文件系统的所有更新操作的路径，客户端的所有写操作都会被记录Edits
3. send_txid文件保存的是一个数字，就是最后一个edits_的数字
4. 每次NameNode启动的时候都会将Fsimage文件读入内存，加载Edits里面的更新操作，保证内存中得元数据信息是最新的、同步的，可以看成NameNode启动的时候就将Fsimage和Edits文件进行了合并

oiv查看Fsimage文件
hdfs oiv -p 文件类型 -i 镜像文件 -o 转换后文件输出路径

cd /opt/module/hadoop-3.1.3/data/dfs/name/current
hdfs oiv -p XML -i fsimage_0000000000000000025 -o /opt/module/hadoop-3.1.3/fsimage.xml

Fsimage 中没有记录块所对应 DataNode，为什么？
在集群启动后，要求 DataNode 上报数据块信息，并间隔一段时间后再次上报。
1
2
3
4
5

oev查看Edits文件
hdfs oev -p 文件类型 -i 编辑日志 -o 转换后文件输出路径

 hdfs oev -p XML -i edits_0000000000000000012-0000000000000000013 -o    	/opt/module/hadoop-
1

3.1.3/edits.xml

cat /opt/module/hadoop-3.1.3/edits.xml

NameNode 如何确定下次开机启动的时候合并哪些 Edits？
	fsiamge_编号，把大于fsimage这个编号的edits_进行合并
1
2
3
4

CheckPoint时间设置

1. 通常情况下。2NN每个一小时执行一次
   hdfs-default.xml

	
	<property>
	 <name>dfs.namenode.checkpoint.periodname>
	 <value>3600svalue>
	property>
1
2
3
4
5

2. 一分钟检查一次操作次数，当操作次数达到 1 百万时，SecondaryNameNode 执行一次

<property>
 <name>dfs.namenode.checkpoint.txnsname>
 <value>1000000value>
<description>操作动作次数description>
property>
<property>
 <name>dfs.namenode.checkpoint.check.periodname>
 <value>60svalue>
<description> 1 分钟检查一次操作次数description>
property>
1
2
3
4
5
6
7
8
9
10

实际开发通常不配置，因为搭建HA，不会用到2NN

DataNode工作机制

1. 一个数据块在DataNode上以文件形式存储在磁盘上，包括两个文件，一个十数据本身，一个是元数据包括数据块的长度、块数据的校验以及时间戳
2. DataNode启动后想NameNode注册，通过后，周期性（6小时）想NameNode上报所有的块信息
   DN向NN汇报当前解读信息的时间间隔，默认6小时

<property>
<name>dfs.blockreport.intervalMsecname>
<value>21600000value>
<description>Determines block reporting interval in 
milliseconds.description>
property>
1
2
3
4
5
6

DN 扫描自己节点块信息列表的时间，默认 6 小时

<property>
<name>dfs.datanode.directoryscan.intervalname>
<value>21600svalue>
<description>Interval in seconds for Datanode to scan data
the disk.
Support multiple time unit suffix(case insensitive), as described
in dfs.heartbeat.interval.
description>
property>
1
2
3
4
5
6
7
8
9

3. 信条是每3秒一次，心跳返回结果带有NameNode给该DataNode的命令如复制块数据到另一台机器，或删除块数据。如果超过10分钟未收到心跳，则认为该节点不可用

数据准确性

CRC校验
1

掉线时限参数设置

需要注意的是 hdfs-site.xml 配置文件中的 heartbeat.recheck.interval 的单位为毫秒，
dfs.heartbeat.interval 的单位为秒。

<property>
 <name>dfs.namenode.heartbeat.recheck-intervalname>
 <value>300000value>
property>
<property>
 <name>dfs.heartbeat.intervalname>
 <value>3value>
property>
1
2
3
4
5
6
7
8

MapReduce

概述

MapReduce是分布式运算的编程框架
优点：
	易于编程。时限业务接口
	良好的扩展性。动态增加服务器，解决计算资源不够的问题
	高容错性。任何一台机器挂掉，可以任务转移
	适合海量数据计算（TB/PB）
缺点：
	不擅长实时计算
	不擅长流逝计算
	不擅长DAG有向无环图计算
	
MapReduce进程
	一个完整的MapReducer程序在分布式运行时有三类进程：
		MrAppMaster：调度
		MapTask：
		ReduceTask
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

序列化

Java 类型 			Hadoop Writable 类型
Boolean 				BooleanWritable
Byte					ByteWritable
Int 					IntWritable
Float 					FloatWritable
Long 					LongWritable
Double					 DoubleWritable
String 					Text
Map 					MapWritable
Array				 ArrayWritable
Null 					NullWritable
1
2
3
4
5
6
7
8
9
10
11

wordcount

    4.0.0

    org.example
    MapReduceDemo
    1.0-SNAPSHOT
    jar

    MapReduceDemo
    http://maven.apache.org

    
        UTF-8
    

    
        
            org.apache.hadoop
            hadoop-client
            3.1.3
        
        
            junit
            junit
            4.12
        
        
            org.slf4j
            slf4j-log4j12
            1.7.30
        
    
    
        
            
                maven-compiler-plugin
                3.6.1
                
                    1.8
                    1.8
                
            
            
                maven-assembly-plugin
                
                    
                        jar-with-dependencies
                    
                
                
                    
                        make-assembly
                        package
                        
                            single
                        
                    
                
            
        
    



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

log4j.properties

log4j.rootLogger=INFO, stdout 
log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n 
log4j.appender.logfile=org.apache.log4j.FileAppender 
log4j.appender.logfile.File=target/spring.log 
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout 
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n


1
2
3
4
5
6
7
8
9
10

package com.atguigu.mapreduce.wordcount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class WordCountMapper extends Mapper<LongWritable, Text, Text,
        IntWritable> {
    Text k = new Text();
    IntWritable v = new IntWritable(1);

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
// 1 获取一行
        String line = value.toString();
// 2 切割
        String[] words = line.split(" ");
// 3 输出
        for (String word : words) {
            k.set(word);
            context.write(k, v);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

package com.atguigu.mapreduce.wordcount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class WordCountReducer extends Reducer<Text, IntWritable, Text,
        IntWritable> {
    int sum;
    IntWritable v = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context
            context) throws IOException, InterruptedException {
// 1 累加求和
        sum = 0;
        for (IntWritable count : values) {
            sum += count.get();
        }
// 2 输出
        v.set(sum);
        context.write(key, v);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

package com.atguigu.mapreduce.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class WordCountDriver {
    public static void main(String[] args) throws IOException,
            ClassNotFoundException, InterruptedException {
// 1 获取配置信息以及获取 job 对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
// 2 关联本 Driver 程序的 jar
        job.setJarByClass(WordCountDriver.class);
// 3 关联 Mapper 和 Reducer 的 jar
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
// 4 设置 Mapper 输出的 kv 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
// 5 设置最终输出 kv 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
// 6 设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 7 提交 job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

打包
测试：

hadoop jar MapReduceDemo-1.0-SNAPSHOT.jar com.atguigu.mapreduce.wordcount.WordCountDriver /user/atguigu/input/word.txt /user/atguigu/output

Hadoop序列化

具体实现 bean 对象序列化步骤如下 7 步。
（1）必须实现 Writable 接口
（2）反序列化时，需要反射调用空参构造函数，所以必须有空参构造
public FlowBean() {
super();
}
（3）重写序列化方法

	@Override
	public void write(DataOutput out) throws IOException {
	out.writeLong(upFlow);
	out.writeLong(downFlow);
	out.writeLong(sumFlow);
	}
1
2
3
4
5
6

（4）重写反序列化方法

		@Override
			public void readFields(DataInput in) throws IOException {
			upFlow = in.readLong();
			downFlow = in.readLong();
			sumFlow = in.readLong();
			}
1
2
3
4
5
6

（5）注意反序列化的顺序和序列化的顺序完全一致
（6）要想把结果显示在文件中，需要重写 toString()，可用"\t"分开，方便后续用。
（7）如果需要将自定义的 bean 放在 key 中传输，则还需要实现 Comparable 接口，因为
MapReduce 框中的 Shuffle 过程要求对 key 必须能排序。详见后面排序案例。

@Override
public int compareTo(FlowBean o) {
// 倒序排列，从大到小
return this.sumFlow > o.getSumFlow() ? -1 : 1;
}
1
2
3
4
5

序列化案例： 统计每一个手机号耗费的总上行流量、总下行流量、总流量
（1）输入数据
（2）输入数据格式：
7 13560436666 120.196.100.99 1116 954 200
id 手机号码 网络 ip 上行流量 下行流量 网络状态码
（3）期望输出数据格式
13560436666 1116 954 2070
手机号码 上行流量 下行流量 总流量

public class FlowBean implements Writable {
    private long upFlow; //上行流量
    private long downFlow; //下行流量
    private long sumFlow; //总流量

    //2 提供无参构造
    public FlowBean() {
    }

    //3 提供三个参数的 getter 和 setter 方法
    public long getUpFlow() {
        return upFlow;
    }

    public void setUpFlow(long upFlow) {
        this.upFlow = upFlow;
    }

    public long getDownFlow() {
        return downFlow;
    }

    public void setDownFlow(long downFlow) {
        this.downFlow = downFlow;
    }

    public long getSumFlow() {
        return sumFlow;
    }

    public void setSumFlow(long sumFlow) {
        this.sumFlow = sumFlow;
    }

    public void setSumFlow() {
        this.sumFlow = this.upFlow + this.downFlow;
    }

    //4 实现序列化和反序列化方法,注意顺序一定要保持一致
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeLong(upFlow);
        dataOutput.writeLong(downFlow);
        dataOutput.writeLong(sumFlow);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.upFlow = dataInput.readLong();
        this.downFlow = dataInput.readLong();
        this.sumFlow = dataInput.readLong();
    }

    //5 重写 ToString
    @Override
    public String toString() {
        return upFlow + "\t" + downFlow + "\t" + sumFlow;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean>
{
 private Text outK = new Text();
 private FlowBean outV = new FlowBean();
 @Override
 protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
 //1 获取一行数据,转成字符串
 String line = value.toString();
 //2 切割数据
 String[] split = line.split("\t");
 //3 抓取我们需要的数据:手机号,上行流量,下行流量
 String phone = split[1];
 String up = split[split.length - 3];
 String down = split[split.length - 2];
 //4 封装 outK outV
 outK.set(phone);outV.setUpFlow(Long.parseLong(up));
  outV.setDownFlow(Long.parseLong(down));
  outV.setSumFlow();
  //5 写出 outK outV
  context.write(outK, outV);
 }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean>
{
 private FlowBean outV = new FlowBean();
 @Override
 protected void reduce(Text key, Iterable<FlowBean> values, Context
context) throws IOException, InterruptedException {
 long totalUp = 0;
 long totalDown = 0;
 //1 遍历 values,将其中的上行流量,下行流量分别累加
 for (FlowBean flowBean : values) {
 totalUp += flowBean.getUpFlow();
 totalDown += flowBean.getDownFlow();
 }
 //2 封装 outKV
 outV.setUpFlow(totalUp);
 outV.setDownFlow(totalDown);
 outV.setSumFlow();
 //3 写出 outK outV
 context.write(key,outV);
 }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

public class FlowDriver {
    public static void main(String[] args) throws IOException,
            ClassNotFoundException, InterruptedException {
        //1 获取 job 对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        //2 关联本 Driver 类
        job.setJarByClass(FlowDriver.class);
        //3 关联 Mapper 和 Reducer
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);

//4 设置 Map 端输出 KV 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

//5 设置程序最终输出的 KV 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

//6 设置程序的输入输出路径
        FileInputFormat.setInputPaths(job, new Path("F:\\bigData\\hadoop\\hadoop-learn-master\\笔记（word版本）\\笔记（word版本）\\phone_data.txt"));
        FileOutputFormat.setOutputPath(job, new Path(".\\output"));

//7 提交 Job
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

InputFormat数据输入

切片与MapTask并行度决定机制

切片的个数决定 MapTask个数
1

• MapTask并行度决定机制
  数据块：Block是HDFS物理上把数据分成一块一块。数据块是HDFS存储数据单位
• 数据切片：数据切片只是在逻辑上对输入进行切片，并不会在磁盘上将其切分成片存储。数据切片是MapReduce程序计算输入数据的单位，一个切片会对应启动一个MapTask

数据切片与MapTask并行度决定机制

TextInputFormat

思考：在运行 MapReduce 程序时，输入的文件格式包括：基于行的日志文件、二进制
格式文件、数据库表等。那么，针对不同的数据类型，MapReduce 是如何读取这些数据的呢？
FileInputFormat 常见的接口实现类包括：TextInputFormat、KeyValueTextInputFormat、
NLineInputFormat、CombineTextInputFormat 和自定义 InputFormat 等。

TextInputFormat 是默认的 FileInputFormat 实现类。按行读取每条记录。键是存储该行在整个文件中的起始字节偏移量， LongWritable 类型。值是这行的内容，不包括任何行终止
符（换行符和回车符），Text 类型。

以下是一个示例，比如，一个分片包含了如下 4 条文本记录。

Rich learning form
Intelligent learning engine
Learning more convenient
From the real demand for more close to the enterprise
每条记录表示为以下键/值对：
(0,Rich learning form)
(20,Intelligent learning engine)
(49,Learning more convenient)
(74,From the real demand for more close to the enterprise)

CombineTextInputFormat 切片机制

框架默认的 TextInputFormat 切片机制是对任务按文件规划切片，不管文件多小，都会
是一个单独的切片，都会交给一个 MapTask，这样如果有大量小文件，就会产生大量的
MapTask，处理效率
极其低下
1）应用场景：
CombineTextInputFormat 用于小文件过多的场景，它可以将多个小文件从逻辑上规划到
一个切片中，这样，多个小文件就可以交给一个 MapTask 处理。
2）虚拟存储切片最大值设置
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。
3）切片机制
生成切片过程包括：虚拟存储过程和切片过程二部分。

实现过程
（1）不做任何处理，运行 1.8 节的 WordCount 案例程序，观察切片个数为 4。
number of splits:4
（2）在 WordcountDriver 中增加如下代码，运行程序，并观察运行的切片个数为 3。
（a）驱动类中添加代码如下：
// 如果不设置 InputFormat，它默认用的是 TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);
//虚拟存储切片最大值设置 4m
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);
（b）运行如果为 3 个切片。
number of splits:3
（3）在 WordcountDriver 中增加如下代码，运行程序，并观察运行的切片个数为 1。
（a）驱动中添加代码如下：
// 如果不设置 InputFormat，它默认用的是 TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);
//虚拟存储切片最大值设置 20m
CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);
（b）运行如果为 1 个切片
number of splits:1

MapReduce工作流程

Partition 分区

问题引出
要求将统计结果按照条件输出到不同文件中（分区）。比如：将统计结果按照手机归属地不同省份输出到不同文件中（分区）

默认分区 是根据key的hashCode对ReduceTasks个数取模得到的。用户没法控制哪个
key存储到哪个分区。

自定义Partitioner步骤
（1）自定义类继承Partitioner，重写getPartition()方法

public class PPartitioner extends Partitioner<Text, FlowBean> {
    @Override
    public int getPartition(Text text, FlowBean flowBean, int i) {
        String phone = text.toString();
        String prePhone = phone.substring(0, 3);
        int partition;
        if ("136".equals(prePhone)) {
            partition = 0;
        } else if ("137".equals(prePhone)) {
            partition = 1;
        } else if ("138".equals(prePhone)) {
            partition = 2;
        } else if ("139".equals(prePhone)) {
            partition = 3;
        } else {
            partition = 4;
        }
        return partition;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

（2）在Job驱动中，设置自定义Partitioner,同时 自定义Partition后，要根据自定义Partitioner的逻辑设置相应数量的ReduceTask

//8 指定自定义分区器
 job.setPartitionerClass(PPartitioner.class);
 //9 同时指定相应数量的 ReduceTask
 job.setNumReduceTasks(5);

1
2
3
4
5
6

分区总结

（1）如果ReduceTask的数量> getPartition的结果数，则会多产生几个空的输出文件part-r-000xx；
（2）如果1，则有一部分分区数据无处安放，会Exception；
（3）如 果ReduceTask的数量=1，则不管MapTask端输出多少个分区文件，最终结果都交给这一个
ReduceTask，最终也就只会产生一个结果文件 part-r-00000；

排序是MapReduce框架中最重要的操作之一。

MapTask和ReduceTask均会对数据按照Key进行排序。该操作属于Hadoop默认行为。任何应用程序中的数据均会默认排序，而不是逻辑上是否需要

默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。
1
2
3
4
5

@Override
public int compareTo(FlowBean bean) {
int result;
// 按照总流量大小，倒序排列
if (this.sumFlow > bean.getSumFlow()) {
result = -1;
}else if (this.sumFlow < bean.getSumFlow()) {
result = 1;
}else {
result = 0;
}
return result;
}
1
2
3
4
5
6
7
8
9
10
11
12
13

大公司 公平调度器
小公司 容量调度器
1
2

 yarn application -list
根据 Application 状态过滤：yarn application -list -appStates （所有状态：ALL、NEW、
1
2

application： yarn logs -applicationId application_1612577921195_0001
container： yarn logs -applicationId application_1612577921195_0001 -containerId container_1612577921195_0001_01_000001
1
2

<property>
<description>The class to use as the resource scheduler.description>
<name>yarn.resourcemanager.scheduler.classname>
<value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capaci
ty.CapacitySchedulervalue>
property>

<property>
<description>Number of threads to handle scheduler 
interface.description>
<name>yarn.resourcemanager.scheduler.client.thread-countname>
<value>8value>
property>

<property>
<description>Enable auto-detection of node capabilities such as
memory and CPU.
description>
<name>yarn.nodemanager.resource.detect-hardware-capabilitiesname>
<value>falsevalue>
property>

<property>
<description>Flag to determine if logical processors(such as
hyperthreads) should be counted as cores. Only applicable on Linux
when yarn.nodemanager.resource.cpu-vcores is set to -1 and
yarn.nodemanager.resource.detect-hardware-capabilities is true.
description>
<name>yarn.nodemanager.resource.count-logical-processors-ascoresname>
<value>falsevalue>
property>

<property>
<description>Multiplier to determine how to convert phyiscal cores to
vcores. This value is used if yarn.nodemanager.resource.cpu-vcores
is set to -1(which implies auto-calculate vcores) and
yarn.nodemanager.resource.detect-hardware-capabilities is set to true. 
The number of vcores will be calculated as number of CPUs * multiplier.
description>
<name>yarn.nodemanager.resource.pcores-vcores-multipliername>
<value>1.0value>
property>

<property>
<description>Amount of physical memory, in MB, that can be allocated 
for containers. If set to -1 and
yarn.nodemanager.resource.detect-hardware-capabilities is true, it is
automatically calculated(in case of Windows and Linux).
In other cases, the default is 8192MB.
description>
<name>yarn.nodemanager.resource.memory-mbname>
<value>4096value>
property>

<property>
<description>Number of vcores that can be allocated
for containers. This is used by the RM scheduler when allocating
resources for containers. This is not used to limit the number of
CPUs used by YARN containers. If it is set to -1 and
yarn.nodemanager.resource.detect-hardware-capabilities is true, it is
automatically determined from the hardware in case of Windows and Linux.
In other cases, number of vcores is 8 by default.description>
<name>yarn.nodemanager.resource.cpu-vcoresname>
<value>4value>
property>

<property>
<description>The minimum allocation for every container request at theRM in MBs. Memory requests lower than this will be set to the value of 
this property. Additionally, a node manager that is configured to have 
less memory than this value will be shut down by the resource manager.
description>
<name>yarn.scheduler.minimum-allocation-mbname>
<value>1024value>
property>

<property>
<description>The maximum allocation for every container request at the 
RM in MBs. Memory requests higher than this will throw an
InvalidResourceRequestException.
description>
<name>yarn.scheduler.maximum-allocation-mbname>
<value>2048value>
property>

<property>
<description>The minimum allocation for every container request at the 
RM in terms of virtual CPU cores. Requests lower than this will be set to 
the value of this property. Additionally, a node manager that is configured 
to have fewer virtual cores than this value will be shut down by the 
resource manager.
description>
<name>yarn.scheduler.minimum-allocation-vcoresname>
<value>1value>
property>

<property>
<description>The maximum allocation for every container request at the 
RM in terms of virtual CPU cores. Requests higher than this will throw an
InvalidResourceRequestException.description>
<name>yarn.scheduler.maximum-allocation-vcoresname>
<value>2value>
property>

<property>
<description>Whether virtual memory limits will be enforced for
containers.description>
<name>yarn.nodemanager.vmem-check-enabledname>
<value>falsevalue>
property>

<property>
<description>Ratio between virtual memory to physical memory when
setting memory limits for containers. Container allocations are
expressed in terms of physical memory, and virtual memory usage is 
allowed to exceed this allocation by this ratio.
description>
<name>yarn.nodemanager.vmem-pmem-rationame>
<value>2.1value>
property>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122