Hadoop是什么？ Hadoop是分布式系统基础架构 主要解决海量数据的存储与计算 单纯来讲Hadoop没多大作用，一般指的都是Hadoop生态圈
Hadoop组成 hdfs 存储 mapreduce 计算 yarn 资源调度 common 公共模块
hdfs架构概述 1）NameNode（nn）：存储文件的元数据，如文件名，文件目录结构，文件属性(生成时间，副本数，文件权限)，以及每个文件的块列表和块所在的DataNode等 2）DataNode（dn）：在本地文件系统存储块数据 3）SecondaryNameNode（2nn）：用来监控Hdfs状态的辅助后台程序，每隔一段时间获取hdfs元数据的快照
yarn架构概述 1）ResourceManager（rm）：处理客户端请求，启动/监控ApplicationMaster、监控NodeManager、资源分配与调度; 2）NodeManager（nm）：单个节点上的资源管理、处理来自ResourceManager的命令、处理来自ApplicationMaster的命令; 3）ApplicationMaster：数据切分、为应用程序申请资源，并分配给内部任务、任务监控与容错; 4）Container：对任务运行环境的抽象，封装了CPU、内存等多维资源以及环境变量、启动命令等任务运行相关的信息。
MapReduce架构概述 MapReduce将计算过程分为两个阶段：Map和Reduce 1）Map阶段并行处理输入数据 2）Reduce阶段对Map结果进行汇总
大数据生态体系 1）Sqoop：是Hadoop与mysql之间传送数据的工具。也就是可以将mysql中的数据导入到hdfs中，也可以将hdfs中的数据导入到mysql中。 2）Flume： 3）Kafka： 4）Storm： 5）Spark： 6）Oozie： 7）Hbase： 8）Hive： 9）R语言： 10）Mahout： 11）Zookeeper：
Hadoop运行模式 本地模式(默认模式)：不需要启用单独进程、直接可以运行，测试和开发时使用。 伪分布式模式：等同于完全分布式，只有一个节点。 完全分布式模式：多个节点一起运行。
熟悉四大配置文件： core hdfs map yarn
ssh无密码登录 ssh 另一台电脑的ip地址 不需要输入密码
集群时间同步 找一台机器，作为时间服务器，其他机器定时和时间服务器同步
配置集群常见错误 1）防火墙没关 2）主机名称配置错误 3）ip地址配置错误 4）ssh没配置好 5）root用户和atguigu用户启动集群不统一 6）配置文件修改不细心 7）没有编译源码 8）jps没生效 9）datanode连接不上namenode
HDFS架构 架构主要由四部分组成、分别为HDFS Client、NameNode、DataNode和SecondaryNameNode。 1）Client：客户端 （1）文件切分：文件上传hdfs的时候，Client将文件切分成一个一个的Block，然后进行存储。 （2）与NameNode交互，获取文件的位置信息 （3）与DataNode交互，读取或写入数据 （4）Client提供一些命令来管理HDFS，比如启动或者关闭HDFS （5）Client可以通过一些命令来访问HDFS
2）NameNode:就是一个Master，它是一个主管、管理者 （1）管理hdfs的名称空间 （2）管理数据块的映射信息 （3）配置副本策略 （4）处理客户端读写请求 3）DataNode：就是Slave。NameNode下达命令，DataNode执行实际的操作。 （1）存储实际的数据块 （2）执行数据块的读/写操作 4）SecondaryNameNode：并非NameNode的热备。当NameNode挂掉的时候，它并不能马上替换掉NameNode并提供服务。 （1）辅助NameNode，分担其工作量。 （2）定期合并fsimage和edits，并推送给NameNode。 （3）在紧急情况下，可辅助恢复NameNode
hdfs文件块大小 hdfs中文件在物理上是分块存储（block），块的大小可以通过配置参数（dfs.blocksize)来规定，默认大小在hadoop2.x版本中是128M，老版本是64M
hdfs基本命令行操作 查看某一路径下的所有文件：hdfs dfs -ls [filePath] 上传文件：hdfs dfs -put [local filePath] [hdfs filePath] 下载文件：hdfs dfs -get [hdfs filePath] [local filePath] 删除文件：hdfs dfs -rm -r [hdfs filePath] 查看某一文件内容：hdfs dfs -cat [filePath]
hdfs命令行操作 hdfs客户端操作 通过api操作hdfs 通过io流操作hdfs
hdfs的数据流 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步） 网络拓扑概念 两个节点在一个本地网络中被称为“彼此近邻”是什么意思？海量数据处理时，主要限制速度是节点之间的数据的传输速度—也就是说带宽。也就是说将两个节点间的带宽作为距离的衡量标准。 节点距离：两个节点到达最近的共同祖先的距离总和。 例如：假设有数据中心的d1 机架r1 中的节点n1.该节点可以表示为 /d1/r1/n1。利用这个标记，这里给出四种距离描述。 distance(/d1/r1/n1, /d1/r1/n1) = 0（同一节点上的进程） distance(/d1/r1/n1, /d1/r1/n2) = 2（同一机架上的不同节点） distance(/d1/r1/n1, /d1/r2/n3) = 4（同一数据中心中不同机架上节点） distance(/d1/r1/n1, /d2/r3/n4) = 6 （不同数据中心中的节点） 可以参考下网络拓扑与Hadoop 机架感知（副本节点选择） hdfs如何控制客户端读取哪个副本节点数据？ 客户端离哪个副本数据最近，hdfs就让哪个节点把数据给客户端。
hdfs读数据流程 1）客户端通过Distributed FileSystem向namenode请求下载文件，namenode通过查询元数据，找到文件块所在的datanode地址。 2）挑选一台datanode（就近原则，然后随机）服务器，请求读取数据 3）datanode开始传输数据给客户端（从磁盘里面读取数据输入流，以packet为单位来做校验） 4）客户端以packet为单位接收，先在本地缓存，然后写入目标文件
NameNode工作机制 1、NameNode&SecondaryNameNode工作机制 1）第一阶段：namenode启动 （1）第一次启动namenode格式化后，创建fsimage和edits文件。如果不是第一次启动，直接加载编辑日志和镜像文件到内存 （2）客户端对元数据进行增删改的请求 （3）namenode记录操作日志，更新滚动日志 （4）namenode在内存中对数据进行增删改查 2）第二阶段：SecondaryNameNode工作 （1）SecondaryNameNode询问namenode是否需要checkpoint。直接带回namenode是否检查结果 （2）SecondaryNameNode请求执行checkpoint （3）namenode滚动正在写的edits日志 （4）将滚动前的编辑日志和镜像文件拷贝到SecondaryNameNode （5）SecondaryNameNode加载编辑日志和镜像文件到内存，并合并 （6）生成新的镜像文件fsimage.chkpoint （7）拷贝fsimage.chkpoint到namenode （8）namenode将fsimage.chkpoint，重新命名成fsimage 2、镜像文件和编辑日志文件 1）namenode被格式化后，将在/opt/module/hadoop/data/tmp/dfs/name/current目录中产生如下文件
edits_0000000000000014232-0000000000000014233 fsimage_0000000000000014233.md5 seen_txid VERSION（1）fsimage文件：hdfs文件系统元数据的一个永久性的检查点，其中包含hdfs文件系统的所有目录和文件idnode的序列化信息。 （2）edits文件：存放hdfs文件系统的所有更新操作的路径，文件系统客户端执行的所有写操作首先会记录到edits文件中 （3）seen_txid文件保存的是一个数字，就是最后一个edits_的数字 （4）每次NameNode启动的时候都会将fsimage文件读入内存，并从00001开始到seen_txid中记录的数字依次执行每个edits里面的更新操作，保证内存中的元数据信息是最新的，同步的，可以看成NameNode启动的时候就将fsimage和edits文件进行了合并。 2）oiv查看fsimage文件,oev查看edits文件
hdfs oiv -p XML -i fsimage_0000000000000014233 -o /opt/module/hadoop/fsimage.xml hdfs oev -p XML -i edits_0000000000000014232-0000000000000014233 -o /opt/module/hadoop/edits.xml3）滚动编辑日志 正常情况下hdfs文件系统有更新操作时，就会滚动编辑日志。也可以用命令强制滚动编辑日志 （1）滚动编辑日志(前提必须启动集群)
hdfs dfsadmin -rollEdits（2)什么时候产生镜像文件 NameNode启动时加载镜像文件和编辑日志 4）NameNode版本号 （1）查看namenode版本号 在/opt/module/hadoop/data/tmp/dfs/name/current这个目录下查看VERSION
[root@s240 current]# cat VERSION #Thu Oct 24 06:49:39 EDT 2019 namespaceID=750910234 clusterID=CID-ca738516-bef0-4ac9-8fec-6ec6d06538cc cTime=0 storageType=NAME_NODE blockpoolID=BP-1703405778-192.168.122.240-1571914179051 layoutVersion=-63（2）namenode版本号具体解释 （2.1）namespaceID在hdfs上，会有多个NameNode，所以不同NameNode的namespaceID是不同的，分别管理一组blockpoolID。 （2.2）clusterID集群id，是全局唯一的。 （2.3）cTime属性标记了namenode存储系统的创建时间，对于刚刚格式化的存储系统，这个属性为0；但是在文件系统升级之后，该值会更新到新的时间戳。 （2.4）storageType属性说明该存储目录包含的是namenode的数据结构。 （2.5）blockpoolID:一个block pool id标识一个block pool，并且并且是跨集群的全局唯一。当一个新的NameSpace被创建的时候（format过程的一部分）会创建并持久化一个唯一的ID。在创建过程构建全局唯一的BlockPoolID，比人为的配置更可靠一些。NN将BlockPoolID持久化到磁盘中，在后续的启动过程中，会在此load并使用。 （2.6）layoutVersion是一个负整数。通常只有hdfs增加新特性时才会更新这个版本号。 5）web端访问SecondaryNameNode端口号 浏览器输入：http://s240:50090 查看SecondaryNameNode信息 6）checkpoint检查时间参数设置 （1）查看hdfs-default.xml文件，通常情况下，SecondaryNameNode每隔一个小时执行一次
<property> <name>dfs.namenode.checkpoint.period</name> <value>3600</value> </property>（2）修改hdfs-site.xml文件。一分钟检查一次操作次数，当操作次数达到100w时，SecondaryNameNode执行一次
<property> <name>dfs.namenode.checkpoint.txns</name> <value>1000000</value> <description>操作动作次数</description> </property> <property> <name>dfs.namenode.checkpoint.check.period</name> <value>60</value> <description> 1 分钟检查一次操作次数</description> </property>7）SecondaryNameNode目录结构 SecondaryNameNode用来监控hdfs状态的辅助后台程序，每隔一段时间获取hdfs元数据的快照。 在/opt/module/hadoop/data/tmp/dfs/namesecondary/current这个目录中查看SecondaryNameNode目录结构
edits_0000000000000014234-0000000000000014235 edits_0000000000000014236-0000000000000014237 fsimage_0000000000000014235 fsimage_0000000000000014235.md5 fsimage_0000000000000014237 fsimage_0000000000000014237.md5 VERSIONSecondaryNameNode的namesecondary/current目录和主namenode的current目录的布局相同。 好处：在主namenode发生故障时(假设没有及时备份数据)，可以从SecondaryNameNode恢复数据。 8）NameNode故障处理方法 NameNode故障后，可以采用如下两种方法恢复数据。 （1）将SecondaryNameNode中数据拷贝到namenode存储数据的目录 （1.1）kill -9 namenode进程 （1.2）删除namenode存储的数据
rm -rf /opt/module/hadoop/data/tmp/dfs/name/*(1.3)拷贝SecondaryNameNode中数据到namenode存储数据目录 （1.4）hadoop-daemon.sh start namenode 重启namenode （2）使用-importCheckpoint 选项启动namenode守护进程，从而将SecondaryNameNode中数据拷贝到namenode目录中。 9）集群安全模式操作 （1） 10）NameNode多目录配置 （1）namenode的本地目录可以配置成多个，切每个目录存放内容相同，增加了可读性。 （2）具体配置如下： NameNode的多目录配置
六、DataNode工作机制 1.DataNode工作机制 1）一个数据块在datanode上以文件形式存储在磁盘上，包括两个文件，一个是数据本身，一个是元数据包括数据块的长度，块数据的校验和，以及时间戳。 2）DataNode启动后向namenode注册，通过后，周期性（1小时）的向namenode上报所有的块信息。 3）心跳是每3秒一次，心跳返回结果带有namenode给该datanode的命令如复制块数据到另一台机器，或删除某个数据块。如果超过10分钟没有收到某个datanode的心跳，则认为该节点不可用。 4）集群运行中可以安全加入和退出一些机器 2.数据完整性 1）当DataNode读取block的时候，它会计算checksum 2）如果计算后的checksum，与block创建时值不一样，说明block已经损坏 3）client读取其他DataNode上的block 4）datanode在其文件创建后周期验证checksum 3.掉线实现参数限制 datanode进程死亡或者网络故障造成datanode无法与namenode通信，namenode不会立即把该节点判定为死亡，要经过一段时间，这段时间暂陈做超时时长。hdfs默认的超时时长为10分钟+30秒。如果定义超时时间为timeout，则超时时长的计算公式为： timeout=2dfs.namenode.heartbeat.recheck-interval+10dfs.heartbeat.interval 而默认的dfs.namenode.heartbeat.recheck-interval=5分钟，dfs.heartbeat.interval默认为3s 需要注意的是hdfs-site.xml配置文件中的heartbeat.recheck.interval的单位为毫秒，dfs.heartbeat.interval的单位为秒。 hadoop datanode 节点超时时间设置 4.DataNode的目录结构 和namenode不同的是，datanode的存储目录是初始阶段自动创建的，不需要额外格式化。 1）在/opt/module/hadoop/tmp/hadoop-root/dfs/data/current这个目录下查看版本号
[root@s241 current]# cat VERSION #Thu Oct 24 06:50:45 EDT 2019 storageID=DS-7d390d55-f970-4f01-81b0-92614473acb9 clusterID=CID-ca738516-bef0-4ac9-8fec-6ec6d06538cc cTime=0 datanodeUuid=89fade81-7e72-46dc-a32f-9932a0cb9974 storageType=DATA_NODE layoutVersion=-56 [root@s241 current]#2）具体解释 （1）storageID：存储ID （2）clusterID：集群id，全局唯一 （3）cTime：属性标记了datanode存储系统的创建时间，对于刚刚格式化的存储系统，这个属性为0；但是在文件系统升级之后，该值会更新到新的时间戳。 （4）datanodeUuid：datanode的唯一识别码 （5）layoutVersion是一个负整数。通常只有hdfs增加新特性时才会更新这个版本号 3）在/opt/module/hadoop/tmp/hadoop-root/dfs/data/current/BP-1703405778-192.168.122.240-1571914179051/current这个目录下查看该数据块的版本号
[root@s241 current]# cat VERSION #Thu Oct 24 06:50:45 EDT 2019 namespaceID=750910234 cTime=0 blockpoolID=BP-1703405778-192.168.122.240-1571914179051 layoutVersion=-56 [root@s241 current]#4）具体解释 （1）namespaceID:是datanode首次访问namenode的时候从namenode处获取的storateID，对每个datanode来说是唯一的。namenode可用这个属性来区分不同的datanode （2）cTime属性标记了datanode存储系统的创建时间，对于刚刚格式化的存储系统，这个属性为0；但是在文件系统升级后，该值会更新到新的时间戳。 （3）blockpoolID:一个block pool id标识一个block pool，并且并且是跨集群的全局唯一。当一个新的NameSpace被创建的时候（format过程的一部分）会创建并持久化一个唯一的ID。在创建过程构建全局唯一的BlockPoolID，比人为的配置更可靠一些。NN将BlockPoolID持久化到磁盘中，在后续的启动过程中，会在此load并使用。 （4）layoutVersion是一个负整数。通常只有hdfs增加新特性时才会更 新这个版本号。 5）服役新数据节点 需求：随着公司业务的增长数据量越来越大，原有的数据节点的容量已经不能满足存储数据的需求，因此需要在原有集群基础上动态添加新的数据节点。 （1）环境准备 （1.1）克隆一台虚拟机 （1.2）修改ip和主机名称 （1.3）修改xcall和xsync文件，增加新增节点的同步ssh （1.4）删除原来的hdfs文件系统留存的文件 /opt/module/hadoop/data (2)服役新节点具体步骤 （2.1）在/opt/module/hadoop/etc/hadoop目录下创建dfs.hosts文件
[root@s244 hadoop]# pwd /opt/module/hadoop/etc/hadoop [root@s244 hadoop]# touch dfs.hosts [root@s244 hadoop]# vi dfs.hosts 添加如下主机名称(包含新服役的节点s244) s241 s242 s244（2.2）在namenode的hdfs-site.xml配置文件中增加dfs.hosts属性
<property> <name>dfs.hosts</name> <value>/opt/module/hadoop/etc/hadoop/dfs.hosts</value> </property>(2.3)刷新namenode
[root@s244 hadoop]# hdfs dfsadmin -refreshNodes Refresh nodes successful(2.4)更新resourcemanager节点
[root@s244 hadoop]# yarn rmadmin -refreshNodes（2.5）在namenode的slaves文件中增加新主机名称 增加s244 （2.6）单独命令启动新的数据节点和节点管理器
[root@s244 hadoop]# hadoop-daemon.sh start datanode [root@s244 hadoop]# yarn-daemon.sh start nodemanger(2.7)在web浏览器上检查是否ok 3）如果数据不均衡，可以用命令实现集群的再平衡 [root@s244 hadoop]# start-balancer.sh 6.退役旧数据节点 怎么添加怎么删除，参照上面即可 7.DataNode多目录配置 1）datanode也可以配置成多个目录，每个目录存储的数据不一样。即，数据不是副本。 2）具体配置如下： 七、hdfs其他功能 1.集群间数据拷贝 1）scp实现两个远程主机之间的文件复制 2）discp实现两个hadoop集群之间的递归数据复制 2.hadoop存档 3.快照管理 4.回收站 八、hdfs ha高可用
一、MapReduce入门 1.MapReduce定义 MapReduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个hadoop集群上。 2.优缺点 1）优点 MapReduce易于编程、高容错、拓展性好、适合PB级以上海量数据的离线处理 2）缺点 MapReduce不擅长做实时计算、流式计算、DAG（有向图）计算。 3.MapReduce核心思想 MapReduce的核心编程思想 1）分布式的运算程序往往需要分成至少2个阶段 2）第一个阶段的map task并发实例，完全并行运行，互补相干。 3）第二阶段的reduce task并发实例互不相干，但是他们的数据依赖于上一个阶段的所有map task并发实例的输出。 4）MapReduce编程模型只能包含一个map阶段和一个reduce阶段，如果用户的业务逻辑非常复杂，那就只能多个map reduce程序串行运行。 4.MapReduce进程 一个完整的mapreduce程序在分布式运行时有三类实例进程： 1）MrAPPMaster：负责整个程序的过程调度及状态协调 2）MapTask：负责map阶段的整个数据处理流程 3）ReduceTask：负责reduce阶段的整个数据处理流程 5.MapReduce编程规范（八股文） 二、Hadoop序列化 1.为什么要序列化 一般来说，“活的”对象只生存在内存里，关机断电就没有了。而且“活的”对象只能由本地的进程使用，不能发送到网络上的另外一台计算机。然而序列化可以存储“活的”对象，可以将“活的”对象发送到远程计算机。 2.什么是序列化 序列化就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储（持久化）和网络传输。 反序列化就是将收到字节序列（或其他数据传输协议）或者是硬盘持久化数据，转化成内存中的对象。 3.为什么不用Java的序列化 Java的序列化是一个重量级序列化框架(Serializable)，一个对象被序列化后，会附带很多额外的信息(如各种校验信息，header，继承体系等)，不便于在网络中高效传输。所以，hadoop自己开发了一套序列化机制（Writable），精简、高效。 4.为什么序列化对Hadoop很重要？ 因为hadoop在集群之间进行通讯或者RPC调用的时候，需要序列化，而且要求序列化要快，且体积要小，占用带宽要小。所以必须理解hadoop的序列化机制。 5.常用数据序列化类型 Java类型 | Hadoop Writable类型 boolean | BooleanWritable byte | ByteWritable int | IntWritable float | FloatWritable long | LongWritable double | DoubleWritable string | Text map | MapWritable array | ArrayWritable 6.自定义bean对象实现序列化接口(Writable) 三、MapReduce框架原理 1.MapReduce工作流程 1）流程示意图 MapReduce框架原理 2）流程详解 上面的流程是整个mapreduce最全工作流程，但是shuffle过程只是从第7步开始到第16步结束，具体shuffle过程详解，如下： （1）map task收集我们的map()方法输出的kv对，放到内存缓冲区中 （2）从内存缓冲区不断一处本地磁盘文件，可能会溢出多个文件 （3）多个溢出文件会被合并成大的溢出文件 （4）在溢出过程中，及合并的过程中，都要调用partitioner进行分区和针对key进行排序 （5）reducetask根据自己的分区号，去各个maptask机器上取相应的结果分区数据 （6）reducetask会渠道同一个分区的来自不同maptask的结果文件，reducetask会将这些文件再进行合并(归并排序) （7）合成大文件后，shuffle的过程也就结束了，后面进入reducetask的逻辑运算过程(从文件中取出一个一个的键值对group，调用用户自定义的reduce()方法) 3）注意 shuffle中的缓冲区大小会影响到mapreduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快。 缓冲区的大小可以通过参数调整，参数：io.sort.mb 默认 100M 2.InputFormat数据输入 1）Job提交流程和切片源码详解 2）FileInputFormat切片机制 （1）FileInputFormat中默认的切片机制： （1.1）简单地按照文件的内容长度进行切片 （1.2）切片大小，默认等于block大小 （1.3）切片时不考虑数据集整体，而是逐个针对每一个文件单独切片 比如待处理数据有两个文件：
file1.txt 320M file2.txt 10M经过FileInputFormat的切片机制运算后，形成的切片信息如下：
file1.txt.split1-- 0~128M file1.txt.split2-- 128~256M file1.txt.split3-- 256~320M file2.txt.split-- 0~10M(2)FileInputFormat切片大小的参数配置 通过分析源码，在FileInputFormat中，集权切片大小的逻辑:Math.max(minSize,Math.min(maxSize,blockSize)); 切片主要由这几个值来运算决定 mapreduce.input.fileinputformat.split.minsize=1 默认值为1 mapreduce.input.fileinputformat.split.maxsize=Long.MAXValue 默认值Long.MAXValue.因此，默认情况下，切片大小=blocksize maxsize(切片最大值)：参数如果调得比blocksize小，则会让切片变小，而且就等于配置的这个参数的值。 minsize(切片最小值)：参数调得比blockSize大，则可以让切片变得比blocksize还大。 （3）获取切片信息API
//根据文件类型获取切片信息 FileSplit inputSplit= （FileSplit）context.getInputSplit(); //获取切片的文件名称 String name=inputSplit.getPath().getName();3)CombineTextInputFormat切片机制 关于大量小文件的优化策略 （1）默认情况下TextInputFormat对任务的切片机制是按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个maptask，这样如果有大量小文件，就会产生大量的maptask，处理效率极其低下。 （2）优化策略 （2.1）最好的办法，在数据处理系统的最前端(预处理/采集)，将小文件先合并成大文件，再上传到hdfs做后续分析。 （2.2）补救措施：如果已经是大量小文件在hdfs中了，可以使用另一种InputFormat来做切片（CombineTextInputFormat），它的切片逻辑跟TextFileInputFormat不同：它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个maptask。 （2.3）优先满足最小切片大小，不超过最大切片大小
CombineTextInputFormat.setMaxInputSplitSize(job,4194304);//4m CombineTextInputFormat.setMinInputSplitSize(job,4194304);//2m举例：0.5m+1m+0.3m+5m=2m+4.8m=2m+4m+0.8m （3）具体实现步骤 //如果不设置InputFormat，它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class) CombineTextInputFormat.setMaxInputSplitSize(job,4194304);//4m CombineTextInputFormat.setMinInputSplitSize(job,2097152);//4m(4)实例操作 。。。。 4）InputFormat接口实现类 MapReduce任务的输入文件一般存储在hdfs里面。输入的文件格式包括：基于行的日志文件、二进制格式文件等。这些文件一般会很大，达到数十GB，甚至更大。那么MapReduce是如何读取这些数据的呢？下面首先学习InputFormat接口。 InputFormat常见的接口实现类包括TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定义InputFormat等。 1）TextInputFormat TextInputFormat是默认的InputFormat。每条记录是一行输入。键是LongWritable类型，存储该行在整个文件中的字节偏移量。值是这行的内容，不包括任何行终止负（换行符和回车符）。 2）KeyValueTextInputFormat 每一行均为一条记录，被分隔符分割为key，value。可以通过在驱动类中设置conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR,"");来者定分隔符。默认分隔符是tab(\t). 3)NLineInputFormat 如果是哟给NLineInputFormat，代表每个map进程处理的InputSplit不再按block块去划分，而是按NLineInputFormat指定的行数N来划分。即输入文件的总行数/N=切片数，如果不整除，切片数=商+1. 5）自定义InputFormat 3.MapTask工作机制 1）并行度决定机制 （1）问题引出 maptask的并行度决定map阶段的任务处理并行度，进而影响到整个job的处理速度。那么，maptask并行任务是否越多越好呢？ （2）MapTask并行度决定机制 一个job的map阶段MapTask并行度(个数),由客户端提交job时的切片个数决定。 2）MapTask工作机制 （1）Read阶段：MapTask通过用户编写的RecordReader，从输入InputSplit中解析出一个个key/value。 （2）Map阶段：该节点主要是将解析出的key/value交给用户编写map()函数处理，并产生一系列新的key/value。 （3）Collect收集阶段：在用户编写map()函数中，当数据处理完成后，一般会调用OutputCollector.collect()输出结果。在该函数内部，它会将生成的key/value分区(调用Partitioner)，并写入一个环形内存缓冲区中。 （4）Spill阶段：即”溢写“，当环形缓冲区满后，MapReduce会将数据写道本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。 溢写阶段详情： 步骤1：利用快速排序算法对缓冲区内的数据进行排序，排序方式是，先按照分区编号partition进行排序，然后按照key进行排序。这样，经过排序后，数据以分区为单位聚集在一起，且同一分区内所有数据按照key有序。 步骤2：按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.out（N表示当前溢写次数）中。如果用户设置了Combiner，则写入文件之前，对每个分区中的数据进行一次聚集操作。 步骤3：将分区数据的元信息写到内存索引数据结构SpillRecord中，其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB，则将内存索引写到文件output/spillN.out.index中。 （5）Combine阶段：当所有数据处理完成后，MapTask对所有临时文件进行一次合并，以确保最终只会生成一个数据文件。 当所有数据处理完后，MapTask会将所有临时文件合并成一个大文件，并保存到文件output/file.out中，同时生成相应的索引文件output/file.out.index。 在进行文件合并过程中，MapTask以分区为单位进行合并。对于某个分区，它将采用多轮递归合并的方式。每轮合并io.sort.factor（默认100）个文件，并将产生的文件重新加入待合并列表中，对文件排序后，重复以上过程，直到最终得到一个大文件。 让每个MapTask最终只生成一个数据文件，可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。 4.Shuffle机制 MapReduce确保每个reducer的输入都是按键排序的。系统执行排序的过程（即将map输出作为输入传给reducer）称为shuffle。 5.ReduceTask工作机制 1）设置ReduceTask并行度(个数) reducetask的并行度同样影响整个job的执行并发度和执行效率，但与maptask的并发数由切片数决定不同，Reducetask数量的决定是可以直接手动设置：
//默认值是1,手动设置为4 job.setNumReduceTasks(4)2)注意 （1）reducetask=0,表示没有reduce阶段，输出文件个数和map个数一致。 （2）reducetask默认值就是1,所以输出文件个数为一个。 （3）如果数据分布不均匀，就有可能在reduce阶段产生数据倾斜 （4）reducetask数量并不是任意设置，还要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只有一个reducetask （5）具体多少个reducetask，需要根据集群性能而定 （6）如果分区数不是1,但是reducetask为1,是否执行分区过程。 答案是：不执行分区过程。因为在maptask的源码中，执行分区的前提是先判断reduceNum个数是否大于1.不大于1肯定不执行。 3）实验：测试reducetask多少合适。 （1）实验环境：1个master节点，16个slave节点：cpu：8ghz，内存2g （2）实验结论： 4）ReduceTask工作机制 (1)Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阀值，则写到磁盘上，否则直接放到内存中。 （2）Merge阶段：在远程拷贝数据的同时，ReduceTask启动了两个后台先从对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。 （3）Sort阶段：按照MapReduce语义，用户编写reduce（）函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现了对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。 （4）Reduce阶段：reduce（）函数将计算结果写到hdfs上。 6.OutputFormat数据输出 1） 7.Join多种应用 1）Reduce join （1）原理： Map端的主要工作：为来自不同表（文件）的key/value对打标签以区别不同来源的记录。然后用连接字段作为key，其余部分和新家的标志作为value，最后进行输出。 Reduce端的主要工作：在reduce端以连接字段作为key的分组已经完成，我们只需要在每一个分组当中将那些来源于不同文件的记录分开，最后进行合并就OK了。 （2）该方法的缺点 这种方式的缺点很明显就是会造成map和reduce端，也就是shuffle阶段出现大量的数据传输，效率很低。 （3）实例操作 2）Map join（Distributedcache 分布式缓存） 8.数据清洗(ETL) 1)概述 在运行核心业务MapReduce程序之前，往往要先对数据进行清洗，清理掉不符合用户要求的数据。清理过程往往只需要运行mapper程序，不需要运行reduce程序。 2）实操案例 9.计数器应用 hadoop为每个作业维护若干内置计数器，以描述多项指标。 例如，某些计数器记录已处理的字节数和记录数，使用户可监控已处理的输入数据量和已产生的输出数据量。 10.MapReduce开发总结 四、Hadoop数据压缩 1.概述 压缩技术能够有效减少底层存储系统hdfs读写字节数。 压缩提高了网络带宽和磁盘空间的效率。 在hadoop下，尤其是数据规模很大和工作负载密集的情况下，使用数据压缩显得非常重要。在这种情况下，I/O操作和网络数据传输要花大量的时间。还有，shuffle与merge过程同样也面临着巨大的I/O压力。 鉴于磁盘I/O和网络带宽是hadoop的宝贵资源，数据压缩对于节省资源、最小化磁盘I/O和网络传输非常有帮助。不过，尽管压缩与解压操作的CPU开销不高，其性能的提升和资源的节省并非没有代价。 如果磁盘I/O和网络带宽影响了MapReduce作业性能，在任意MapReduce阶段启用压缩都可以改善端到端处理时间并减少I/O和网络流量。 压缩MapReduce的一种优化策略：通过压缩编码对Mapper或者Reducer的输出进行压缩，以减少磁盘I/O，提高MR程序运行速度(但相应增加了cpu运算负担)。 注意：压缩特性运用得当能提高性能，但运用不当也可能降低性能。 基本原则： （1）运算密集型的job，少用压缩 （2）IO密集型的job，多用压缩 2.MR支持的压缩编码 3.压缩方式选择 4.压缩位置选择 5.压缩配置参数 6.压缩实战 五、Yarn 1.Hadoop1.x和Hadoop2.x架构区别 在Hadoop1.x时代，Hadoop中的MapReduce同时处理业务逻辑运算和资源的调度，耦合性较大。 在2.X时代，增加了Yarn。Yarn只负责资源的调度，MapReduce只负责运算。 2.Yarn概述 Yarn是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台，而MapReduce等运算程序则相当于运行于操作系统之上的应用程序。 3.Yarn基本架构 Yarn主要由ResourceManager、NodeManager、ApplicationMaster和Container等组件构成。 4.Yarn工作机制 1）Yarn运行机制 2）工作机制详解 5.作业提交全过程 6.资源调度器 目前，Hadoop作业调度器主要有三种：FIFO、Capacity Scheduler和Fair Scheduler。Hadoop2.7.2默认的资源调度器是Capacity Scheduler。 7.任务的推测执行 六、hadoop企业优化 1.MapReduce跑的慢的原因 MapReduce程序效率的瓶颈在于两点： 1）计算机性能 cpu、内存、磁盘健康、网络 2）I/O操作优化 （1）数据倾斜 （2）map和reduce数设置不合理 （3）map运行时间太长，导致reduce等待过久 （4）小文件过多 （5）大量的不可分块的超大文件 （6）spill次数过多 （7）merge次数过多等 2.MapReduce优化方法 3.hdfs小文件优化方法 七、MapReduce实战 1.WordCount案例 2.流量汇总案例 3.辅助排序和二次排序案例 4.小文件处理案例 5.过滤日志及自定义日志输出路径案例 6.MapReduce中多表合并案例 7.日志清洗案例 8.倒排索引案例（多job串联） 9.找博客共同好友案例 10.压缩/解压缩案例 八、常见错误 1.导包容易出错。尤其Text和CombineTextInputFormat 2.Mapper中第一个输入的参数必须是LongWritable或者NullWritable，不可以是IntWritable报的错误是类型转换异常。 3.java.lang.Exception:java.io.IOException:Illegal partition for 13926435656(4),说明partition和reducetask个数没对上，调整reducetask个数。 4.如果分区数不是1,但是reducetask为1,是否执行分区过程。答案是：不执行分区过程。因为在maptask的源码中，执行分区的前提是先判断reduceNum个数是否大于1.不大于1肯定不执行。 5.在Windows环境编译的jar包导入到linux环境中运行， hadoop jar wc.jar com.atguigu.mapreduce.wordcount.WordCountDriver /user/atguigu/ /user/atguigu/output 报如下错误：Exception in thread “main” java.lang.UnsupportedClassVersionError:com/atguigu/mapreduce/wordcount/WordCountDriver: Unsupported major.minor version 52.0 原因是Windows环境用的是jdk1.7,linux环境用的jdk1.8 解决方案：同一jdk版本 6.缓存pd.txt小文件案例中，报找不到pd.txt文件 原因：大部分为路径书写错误。还有就是要检查pd.txt.txt的问题。还有个别电脑写相对路径找不到pd.txt，可以修改为绝对路径。 7.报类型转换异常。 通常都在驱动函数中设置map输出和最终输出时编写错误。 Map输出的key如果没有排序，也会报类型转换异常。 8.集群中运行wc.jar时出现了无法获得输入文件。 原因：wordcount案例的输入文件不能放用hdfs集群的根目录。 9.出现了如下相关异常 10.自定义outputformat时，注意在recordWriter中的close方法必须关闭流资源。否则输出的文件内容中数据为空。
