什么是Hadoop
Hadoop是一个由Apache基金会所开发的分布式系统基础架构, 是一个存储系统+计算框架的软件框架。主要解决海量数据存储与计算的问题,整个Hadoop的核心HDFS和MapReduce。HDFS为海量数据提供了存储,而MapReduce为海量数据提供了计算框架。
说说Hadoop生态圈
首先,hadoop生态系统,意思就是以hadoop为平台的各种应用框架,相互兼容,组成了一个独立的应用体系,也可以称之为生态圈。
首先是JDK,java语言的软件开发工具包,是Hadoop最基本的一个服务。HDFS(hadoop分布式文件系统),mapreduce(分布式计算框架),其次是zookeeper(分布式协作服务),解决分布式环境下的数据管理问题:统一命名,状态同步,集群管理,配置同步等。flume(日志收集工具),cloudera提供的开源的日志收集系统,具有分布式,高可靠,高容错,易于定制和扩展的特点。Hive, 基于hadoop的数据仓库。Storm,内存计算框架。可以通过网络直接把数据导入到内存里,直接在内存里计算,流式处理。还有hbase,这是一个面向列的分布式的开源数据库,hbase采用了bigtable的数据模型。Kafka, 这是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据。还有redis,c语言编写的,一个内存数据库。sqoop(数据同步工具)。Impala,一个新的查询系统,非常快,可以查询到存储在hdoop上hdfs和hbase上pb级的大数据。
说说MapReduce的流程和shuffle过程
MapReduce,首先分为Map端,和Reduce端。首先我们有一个待处理的文件,包含了文件的路径,名称,内容和大小。我们有一个客户端,想要跑Mapreduce的程序,我们要把文件提交到集群上,会有一个submit()的方法,在submit之前,获取一下待处理文本的信息,根据参数配置形成一个动态任务的分配规划,比如把文件分成一些数据块,然后提交。提交的地方是yarn, 是集群资源调度的一个组件,提交的信息也包含job.split, wc.jar, job.xml。提交到resourcemanager上,它会根据个数和大小计算,计算出maptask的数量,然后启动。
maptask会调用一个默认的textinputformat,它再会去调用inputformat,它再去调用recorderreader, 调用reader方法,将数据一行一行读取,根据rr将数据抽成一行一行key,value的形式。key就是行首字母的偏移量。value就是一行数据,以k,v的方法会到这个mapper方法里面去。
map()根据传进来的kv进行逻辑运算,再write()出去,标记好分区,发送到环形缓冲区当中,环形缓冲区的默认大小是100M, 达到80%的阈值,会溢写。溢写会产生大量文件。溢写之前会分区和排序,排序的规则是将k进行字典序排序,快排,分区的规则就是将key取哈希值,哈希值除以reduce task的数量,余几就放到哪个分区里面去。分区内的文件也是有序的,溢写产生的这些大量文件,会再进行merge操作,进行归并排序,默认把10个溢写文件合并成大文件,也可以对数据进行combiner的操作,combiner的结果也不能对最终结果产生影响,等所有maptask完成操作后,启动reduce task。
reducetask会发取拉取的线程到map端拉取数据,一次发送十个,数据先加载到内存里,内存不够就写到磁盘里去,等数据都拉过来了,再进行一次归并排序,归并完的文件进行一次分组的操作,将数据以组为单位发送到reduce()中,做完逻辑运算后,最终会调用outputformat,在调用recorderwriter,将数据以kv形式进行输出。
Hadoop端口
dfs.namenode.http-address:50070
SecondaryNameNode 辅助名称节点端口号:50090
dfs.datanode.address:50010
fs.defaultFS:8020 或者 9000 文件系统做交互的端口
yarn.resourcemanager.webapp.address:8088
历史服务器 web 访问端口:19888
配置文件
core-site,xml
hdfs.site.xml
mapred-site.xml
yarn-site.xml
hadoop-env.sh
yarn-env.sh
mapred-env.sh
slaves
MapReduce的两表join
Mapjoin, 有两个表,一个非常大,一个非常小,小表直接放在内存中,把小表复制成多份,每一个maptask里都放一个,然后去扫描大表,大表的每一个记录的key,value,与小表做join。
还有一种是reduce join,是map阶段,map函数同时读取两个文件,对每一条读取到的数据,打上一个标签,经过suffle分组后,在reduce端会得到一个kv的List,遍历完后,比如把a的放一个list,b的放一个List,去做乘积,做join。但是,suffle阶段传输和排序性能很低的,而且做乘积很耗内粗内存,很容易导致内存溢出。
讲讲MapReduce的优化
小文件的优化:不适合小文件。存储方面,hadoop存储每个文件,都在namenode上记录了元数据信息,namenode中,每个存储的元数据大概150字节左右,小文件越多,文件记录越多。读取来说,增加磁盘寻址的次数,降低性能。计算方面来说,map一般处理一个切片或小文件,如果map启动的时间比处理数据的时间都长,性能有问题,如果map很多,中间溢写产生的临时数据也会很多,reduce拉数据也会增加IO.怎么解决,不存小文件,读取的时候,combinefileinputformat,进行合并,然后计算。
数据倾斜:mapreduce是并行处理的,处理的时间肯定是所有任务里最慢的那一个。怎么产生数据倾斜,分区的时候我们是根据k取哈希值,然后除以reducetask,余几就放到哪个分区,有可能其中一个k匹配了很多的数据,都放到一个reduce里,导致数据倾斜。解决办法:自定义分区。加reduce处理的内存,增加reduce的个数,或使用推测执行,以空间换时间进行优化。
JVM重用:每读取一个小文件,就会读取一个mptask,启动消耗时间,一个 map 运行一个 jvm,重用的话,在一个 map 在 jvm 上运行完毕后,jvm 继续运行其他 map。
map端调参可以优化,减少环形缓冲区flush的次数,如调大环形缓冲区的大小,调大阈值,对Map输出的数据进行压缩,
reduce端尽量让所有数据都在内存里,内存大点。
集群调优核心思路:在网络带宽,磁盘IO是瓶颈的前提下,能不使用IO和网络,就不使用,在必须使用的情况下,能少使用网络就少使用。