Hadoop

MR on Yarn、文件格式、文件压缩

Posted by Jackson on 2017-08-24

ResourceManager：
application Manager 应用程序管理器
resource schedule 资源调度器
container

MR on Yarn 的流程

1.用户向Yarn提交应用程序(job、application)，jar包、sql等其中包含了applicationMaster程序、启动application Master的命令等。
2.RM为该job分配第一个container，用来运行applicationMaster，
3.applicationMaster向applicationManager注册，这样就可以在RM的WEB界面查看此job的运行状态
4.ApplicationMaster采用轮询的方式通过RPC协议向RM申请和领取资源

5.ApplicationMaster获取到资源后与对应的NameNode进行通信，要求启动任务
6.NameNode为任务设置好运行环境， 将任务的启动命令写在一个脚本里面，并且通过该脚本启动task
7.每个task通过rpc协议向ApplicationMaster汇报自己的状态和进度，方便ApplicationMaster随时掌控每个task的运行状态。从而在task运行失败的时候重启任务
8.ApplicationMaster向ApplicationManager注销且关闭自己

使用hadoop自带jar包跑mr任务

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
[hadoop@bigdata01 hadoop]$ hadoop jar ./share/hadoop/mapreduce2/hadoop-mapreduce-examples-2.6.0-cdh5.16.2.jar wordcount /wordcount/input /wordcount/output2
19/12/09 20:46:54 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
19/12/09 20:46:54 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
19/12/09 20:46:55 INFO mapreduce.JobSubmissionFiles: Permissions on staging directory /tmp/hadoop-yarn/staging/hadoop/.staging are incorrect: rwxr-xr-x. Fixing permissions to correct value rwx------
19/12/09 20:46:55 INFO input.FileInputFormat: Total input paths to process : 1
19/12/09 20:46:55 INFO mapreduce.JobSubmitter: number of splits:1
19/12/09 20:46:55 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1575895378954_0001
19/12/09 20:46:56 INFO impl.YarnClientImpl: Submitted application application_1575895378954_0001
19/12/09 20:46:56 INFO mapreduce.Job: The url to track the job: http://bigdata01:8088/proxy/application_1575895378954_0001/
19/12/09 20:46:56 INFO mapreduce.Job: Running job: job_1575895378954_0001
19/12/09 20:47:05 INFO mapreduce.Job: Job job_1575895378954_0001 running in uber mode : false
19/12/09 20:47:05 INFO mapreduce.Job:  map 0% reduce 0%
19/12/09 20:47:10 INFO mapreduce.Job:  map 100% reduce 0%
19/12/09 20:47:16 INFO mapreduce.Job:  map 100% reduce 100%
19/12/09 20:47:16 INFO mapreduce.Job: Job job_1575895378954_0001 completed successfully
19/12/09 20:47:16 INFO mapreduce.Job: Counters: 49
	File System Counters
		FILE: Number of bytes read=114
		FILE: Number of bytes written=286251
		FILE: Number of read operations=0
		FILE: Number of large read operations=0
		FILE: Number of write operations=0
		HDFS: Number of bytes read=221
		HDFS: Number of bytes written=72
		HDFS: Number of read operations=6
		HDFS: Number of large read operations=0
		HDFS: Number of write operations=2
	Job Counters 
		Launched map tasks=1
		Launched reduce tasks=1
		。。。
		Total megabyte-milliseconds taken by all reduce tasks=3035136
	Map-Reduce Framework
		Map input records=3
		Map output records=18
		Map output bytes=180
		Map output materialized bytes=114
		Input split bytes=111
		Combine input records=18
		Combine output records=9
		Reduce input groups=9
		Reduce shuffle bytes=114
		Reduce input records=9
		Reduce output records=9
		Spilled Records=18
		Shuffled Maps =1
		Failed Shuffles=0
		Merged Map outputs=1
		GC time elapsed (ms)=104
		CPU time spent (ms)=1120
		Physical memory (bytes) snapshot=325328896
		Virtual memory (bytes) snapshot=5455265792
		Total committed heap usage (bytes)=226627584
	Shuffle Errors
		BAD_ID=0
		CONNECTION=0
		IO_ERROR=0
		WRONG_LENGTH=0
		WRONG_MAP=0
		WRONG_REDUCE=0
	File Input Format Counters 
		Bytes Read=110
	File Output Format Counters 
		Bytes Written=72

[hadoop@bigdata01 hadoop]$ hadoop fs -cat /wordcount/output2/part-r-00000
19/12/09 20:50:01 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
flink	3
flume	2
hadoop	3
hbase	2
hive	2
impala	1
kafka	1
kudu	2
spark	2

MapTask的个数和ReduceTask的个数

在hadoop官网的mapred-default.xml 中有相关的配置参数：
mapreduce.job.maps 默认的个数是2
mapreduce.job.reduces 默认的个数是1

mapTask的个数实际上是取决于split的个数、一般情况下split的个数等于block的个数，

在生产上面尽量控制一个文件的大小稍微小于一个blocksize的大小
假设我们有三个输入文件，分别是3M、65M、127M，假设block的大小是64M，那么这里会将3M的文件划分成为一个block块，将65M的文件划分成为2个文件，将127M的文件划分成为两个文件，此时我们进行运算的时候会有5个map任务执行，但是每个mapTask执行的数据量的大小是不均匀的，那么这个是可以进行优化的。

文件类型和格式

各种文件格式的优缺点：
ORC：
ORC是列式存储，有多种文件压缩方式，并且有着很高的压缩比。
文件是可切分（Split）的。因此，在Hive中使用ORC作为表的文件存储格式，不仅节省HDFS存储资源，查询任务的输入数据量减少，使用的MapTask也就减少了。
提供了多种索引，row group index、bloom filter index。
ORC可以支持复杂的数据结构（比如Map等）

在生产上面：我们一般使用的是textfile orc 或者parquet格式的文件
压缩的格式一般有gzip snappy lzo

文件压缩格式

压缩的好处和坏处

好处：

  • 1.减少存储磁盘空间
  • 2.降低IO(网络的IO和磁盘的IO)
  • 3.加快数据在磁盘和网络中的传输速度，从而提高系统的处理速度

坏处

  • 由于使用数据时，需要先将数据解压，加重CPU负荷

压缩格式：
avatar

压缩比例：
avatar

压缩时间:
avatar

压缩格式 优点 缺点
gzip 压缩比在四种压缩方式中较高；hadoop本身支持，在应用中处理gzip格式的文件就和直接处理文本一样；有hadoop native库；大部分linux系统都自带gzip命令，使用方便 不支持split
lzo 压缩/解压速度也比较快，合理的压缩率；支持split，是hadoop中最流行的压缩格式；支持hadoop native库；需要在linux系统下自行安装lzop命令，使用方便 压缩率比gzip要低；hadoop本身不支持，需要安装；lzo虽然支持split，但需要对lzo文件建索引，否则hadoop也是会把lzo文件看成一个普通文件（为了支持split需要建索引，需要指定inputformat为lzo格式）
snappy 压缩速度快；支持hadoop native库 不支持split；压缩比低；hadoop本身不支持，需要安装；linux系统下没有对应的命令d. bzip2
bzip2 支持split；具有很高的压缩率，比gzip压缩率都高；hadoop本身支持，但不支持native；在linux系统下自带bzip2命令，使用方便 压缩/解压速度慢；不支持native

总结

不同的场景选择不同的压缩方式，肯定没有一个一劳永逸的方法，如果选择高压缩比，那么对于cpu的性能要求要高，同时压缩、解压时间耗费也多；选择压缩比低的，对于磁盘io、网络io的时间要多，空间占据要多；对于支持分割的，可以实现并行处理。

应用场景

在生产上面当我们的使用场景不同的时候，可以采用不同的文件格式和压缩格式
hive：orc/parquet + bzip2
hbase:hfile + snappy

MapReduce的shuffle

avatar

map端的shuffle

  • map端会处理出入数据并产生中间结果，这个中间结果会写到本地磁盘，而不是HDFS。每个map的输出会先写到内存缓冲区中，当写入的数据达到设定的阈值时，系统将会启动一个线程将缓冲区的数据写到磁盘，这个过程叫做spill(溢写)。
  • 在spill之前，会先进行两次排序，首先根据数据所属的partition进行排序，然后每个partition中的数据再按key来排序，partition的目的是将记录划分到不同的reduce上，以期望能达到负载均衡，以后的reduce就会根据partition来读取自己对应的数据。接着运行combiner(如果设置了的话)，combiner的本质也是一个reduce，其目的是对将要写入到磁盘的文件先进行一次处理，这样，写入到磁盘的数据量就会减少。最后将数据写到本地磁盘产生spill文件(spill文件保存在{mapred.local.dir}指定的目录中，map任务结束后就会被删除)。
  • 最后，每个map任务可能产生多个spill文件，在每个map任务完成前，会通过多路归并算法将这些spill文件合并成一个文件。至此，map的shuffle过程就结束了。
    reduce端的shuffle

reduce端的shuffle主要包括三个阶段，copy、sort(merge)和reduce

  • 首先将map端产生的输出文件拷贝到reduce端，但每个reduce如何知道自己应该处理哪些数据呢？因为map端进行partition的时候，实际上就相当于指定了每个reduce要处理的数据(partition就对应了reduce)，所以reduce在拷贝的数据的时候只需拷贝与自己对应的partition中的数据即可。每个reduce会处理一个或多个partiton，但需要先将自己对应的partition中的数据从每个map的输出结果中拷贝出来。
  • 接下来就是sort阶段，也称为merge阶段，因为这个阶段的主要工作是执行了归并排序。从map端拷贝到reduce端的数据都是有序的，所以很适合归并排序。最终在reduce端生产一个较大的文件作为reduce的输入。
  • 最后就是reduce阶段了，在这个过程中产生最终的输出结果，并将其写到HDFS上。

将map的输出作为reduce的输入的过程就是shuffle了，这个是mapreduce优化的重点地方。Shuffle一开始就是map阶段做输出操作，一般mapreduce计算的都是海量数据，map输出时候不可能把所有文件都放到内存操作，因此map写入磁盘的过程十分的复杂，更何况map输出时候要对结果进行排序，内存开销是很大的，map在做输出时候会在内存里开启一个环形内存缓冲区，这个缓冲区专门用来输出的，默认大小是100mb，并且在配置文件里为这个缓冲区设定了一个阀值，默认是0.80（这个大小和阀值都是可以在配置文件里进行配置的），同时map还会为输出操作启动一个守护线程，如果缓冲区的内存达到了阀值的80%时候，这个守护线程就会把内容写到磁盘上，这个过程叫spill，另外的20%内存可以继续写入要写进磁盘的数据，写入磁盘和写入内存操作是互不干扰的，如果缓存区被撑满了，那么map就会阻塞写入内存的操作，让写入磁盘操作完成后再继续执行写入内存操作，前面我讲到写入磁盘前会有个排序操作，这个是在写入磁盘操作时候进行，不是在写入内存时候进行的，如果我们定义了combiner函数，那么排序前还会执行combiner操作。每次spill操作也就是写入磁盘操作时候就会写一个溢出文件，也就是说在做map输出有几次spill就会产生多少个溢出文件，等map输出全部做完后，map会合并这些输出文件。这个过程里还会有一个Partitioner操作，对于这个操作很多人都很迷糊，其实Partitioner操作和map阶段的输入分片（Input split）很像，一个Partitioner对应一个reduce作业，如果我们mapreduce操作只有一个reduce操作，那么Partitioner就只有一个，如果我们有多个reduce操作，那么Partitioner对应的就会有多个，Partitioner因此就是reduce的输入分片，这个程序员可以编程控制，主要是根据实际key和value的值，根据实际业务类型或者为了更好的reduce负载均衡要求进行，这是提高reduce效率的一个关键所在。到了reduce阶段就是合并map输出文件了，Partitioner会找到对应的map输出文件，然后进行复制操作，复制操作时reduce会开启几个复制线程，这些线程默认个数是5个，程序员也可以在配置文件更改复制线程的个数，这个复制过程和map写入磁盘过程类似，也有阀值和内存大小，阀值一样可以在配置文件里配置，而内存大小是直接使用reduce的tasktracker的内存大小，复制时候reduce还会进行排序操作和合并文件操作，这些操作完了就会进行reduce计算了。

WordCount图解

avatar




FEATURED TAGS
Hadoop
FRIENDS