Hadoop基本概念及MapReduce编程模型

1、NameNode

       NameNode是一个中心服务器，单一节点（简化系统的设计和实现），负责管理文件系统的名字空间（namespace）以及客户端对文件的访问。

       NameNode负责文件元数据的操作，DataNode负责处理文件内容的读写请求，跟文件内容相关的数据流不经过NameNode，只会询问它跟哪个DataNode联系，否则NameNode会成为系统的瓶颈。

       副本存放在哪些DataNode上由NameNode来控制，根据全局情况做出块放置决定，读取文件时NameNode尽量让用户先读取最近的副本，降低带块消耗和读取时延。

       NameNode全权管理数据块的复制，它周期性地从集群中的每个DataNode接收心跳信号和块状态报告（Blockreport）。接收到心跳信号意味着该DataNode节点工作正常。块状态报告包含了一个该DataNode上所有数据块的列表。

2、DataNode

      一个数据块在DataNode以文件存储在磁盘上，包括两个文件，一个是数据本身，一个是元数据，包括数据块的长度，块数据的校验和，以及时间戳。

      DataNode启动后向NameNode注册，通过后，周期性（1小时）的向NameNode上报所有的块信息。

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

      集群运行中可以安全加入和退出一些机器。

3、文件

      文件切分成块（默认大小128M），以块为单位，每个块有多个副本存储在不同的机器上，副本数可在文件生成时指定（默认3）。

      NameNode是主节点，存储文件的元数据，如文件名，文件目录结构，文件属性（生成时间，副本数，文件权限），以及每个文件的块列表以及块所在的DataNode等等。

      DataNode在本地文件系统存储文件块数据，以及块数据的校验和。

      可以创建、删除、移动或重命名文件，当文件创建、写入和关闭之后不能修改文件内容。

4、YARN服务组件

      YARN总体上仍然是Master/Slave结构，在整个资源管理框架中，ResourceManager为Master，NodeManager为Slave。

      ResourceManager负责对各个NodeManager上的资源进行统一管理和调度。

      当用户提交一个应用程序时，需要提供一个用以跟踪和管理这个程序的ApplicationMaster，它负责向ResourceManager申请资源，并要求NodeManager启动可以占用一定资源的任务。

      由于不同的ApplicationMaster被分布到不同的节点上，因此它们之间不会相互影响。

5、ResourceManager

      全局的资源管理器，整个集群只有一个，负责集群资源的统一管理和调度分配。

      功能：处理客户端请求，启动/监控ApplicationMaster，监控NodeManager，资源分配与调度。

6、NodeManager

      整个集群有多个，负责单节点资源管理和使用。

      功能：单个节点上的资源管理和任务管理，处理来自ResourceManager的命令和来自ApplicationMaster的命令。

      NodeManager管理抽象容器，这些容器代表着可供一个特定应用程序使用的针对每个节点的资源。

      定时地向ResourceManager汇报本节点上的资源使用情况和各个Container的运行状态。

7、Application Master

      管理一个在YARN内运行的应用程序的每个实例。

      功能：负责数据切分，为应用程序申请资源，并进一步分配给内部任务，任务监控与容错。

      负责协调来自ResourceManager的资源，并通过NodeManager监视容器的执行和资源使用（CPU、内存等的资源分配）。

8、Container

      YARN中的资源抽象，封装某个节点上多维度资源，如内存、CPU、磁盘、网络等，当Application Master向ResourceManager申请资源时，ResourceManager向Application Master返回的资源便是用Container表示的。

      YARN会为每个任务分配一个Container，且该任务只能使用该Container中描述的资源。

      功能：负责对任务运行环境的抽象，描述一系列信息，任务运行资源（节点、内存、CPU），任务启动命令和任务运行环境。

9、YARN资源管理

      资源调度和资源隔离是YARN作为一个资源管理系统最重要和最基础的两个功能。资源调度由ResourceManager完成，而资源隔离由各个NodeManager实现。

      ResourceManager将某个NodeManager上资源分配给任务（这就是所谓的“资源调度”）后，NodeManager需按照要求为任务提供相应的资源，甚至保证这些资源应具有独占性，为任务运行提供基础的保证，这就是所谓的资源隔离。

      当谈及到资源时，我们通常指内存、CPU和IO三种资源。Hadoop YARN同时支持内存和CPU两种资源的调度。

      内存资源的多少会决定任务的生死，如果内存不够，任务可能会运行失败；相比之下，CPU资源则不同，它只会决定任务运行的快慢，不会对生死产生影响。

10、HDFS启动过程

     （1）注册（心跳）、块的状态报告；

     （2）namenode启动的时候会有一个30秒的等待过程；

     （3）datanode会周期性的向namenode发送心跳（每3秒），namenode接收心跳信号会给datanode进行一个反馈，反馈中可能会附带一些指令；

     （4）在启动的过程中向namenode汇报块的报告；

     （5）读取fsimage和edits文件，获取元数据信息；

     （6）安全模式，namenode在启动的过程中（30秒等待）是处于安全模式的，安全模式是只读的。

              $ bin/hdfs dfsadmin -safemode get

              Safe mode is OFF

              $ bin/hdfs dfsadmin -safemode enter

              Safe mode is ON

              $ bin/hdfs dfsadmin -safemode leave

              Safe mode is OFF

     （7）HDFS上传下载命令：

              bin/hdfs dfs -put etc/hadoop/core-site.xml /user/beifeng/temp/conf

              bin/hdfs dfs -getmerge /user/beifeng/temp/conf/hdfs-site.xml /user/beifeng/temp/conf/core-site.xml /home/beifeng/beifeng-site.xml

11、HDFS Java API

      （1）获取配置信息。Configuration configuration = new Configuration();

      （2）获取文件系统，传递配置信息。FileSystem fileSystem = FileSystem.get(configuration);

      （3）读取文件内容，相当于bin/hdfs dfs -text 操作。

      （4）对于文件的读写，流操作。

      （5）HDFS Java API用于解决存储问题。

12、MapReduce编程模型

     （1）MapReduce用于解决计算问题，处理的对象都是键值对<key, value>。

     （2）MapReduce计算框架特点：

               >>计算过程分为两个阶段：Map和Reduce，Map阶段并行处理输入数据，Reduce阶段对Map结果进行汇总。

               >>Shuffle链接Map和Reduce两个阶段，Map Task 将数据写到本地磁盘，Reduce Task 从每个Map Task上读取一份数据。

               >>仅适合离线批处理，具有很好的容错性和扩展性，适合简单的批处理任务。

               >>缺点明显：启动开销大，过多使用磁盘导致效率地下等。

     （3）一个mapreduce程序分为5块：

             input -> map() -> shuffle -> reduce() -> output

     （4）举例：单词统计wordcount程序

             hadoop mapreduce       <0, hadoop mapreduce>

             hadoop yarn

             hadoop spark

             mapreduce hive

             hadoop hello

             将每一行变成<key, value>，key是行偏移量（数字），value是这一行的内容。

             第一步：获取每一行的值；

             第二步：分割单词；每个单词出现一次就记录一次-><hadoop, 1><mapreduce, 1>->map过程，map的<key, value>；

             第三步：将相同key的value相加在一起，<hadoop, list(1,1,1,1)>->reduce过程，reduce的<key, value>。

     （5）map和reduce都要继承于它们的基类Mapper和Reducer。

              mapreduce程序分为Mapper区块、Reducer区块和Driver区块。

     （6）在yarn上运行wordcount：

              在eclipse中将.java文件导出成jar包，位置为/opt/modules/hadoop-2.5.0/jars/mr-wc.jar

              $ bin/yarn jar jars/mr-wc.jar /user/beifeng/wordcount/input /user/beifeng/wordcount/output

              $ bin/yarn jar jars/mr-wc.jar com.ibeifeng.mapreduce.WordCountMapReduce /user/beifeng/wordcount/input /user/beifeng/wordcount/output

              查看wordcount统计结果：$ bin/hdfs dfs -text /user/beifeng/wordcount/output/part*

              hadoop  4

              hello   1

              hive    1

              mapreduce       2

              spark   1

              yarn    1

              默认情况下一个nodemanager块上执行一个map任务，一个完整的job任务就打一个jar包。

13、mapreduce shuffle

       Shuffle：洗牌或弄乱。

       Collection.shuffle(List)：随机地打乱参数list里的元素顺序。

       MapReduce里的Shuffle：描述着数据从map task输出到reduce task输入的这段过程。

       发生阶段：map() -> shuffle -> reduce()

       数据输入<key, value>

            <0, hadoop spark>

                         |

                     map()

                         |

           <hadoop, 1><spark, 1>

                         |

                 map output

            map端shuffle 

                        >> 将结果输出到内存中，默认大小100MB，可以自定义

                        >>将输出结果放到环形缓冲区，当内存使用空间达到80%，80MB（可以自定义）

                        >>进行spill溢写操作，溢写到磁盘，本地磁盘的某个目录中

                              >>>分区partition，决定了map输出的数据被哪个reduce任务进行处理

                              >>>排序sort，对分区阶段中的数据进行排序，在内存中排序 > 比较

                              >>>溢写spill，溢写到本地磁盘某个目录中，形成一个溢写文件

                              >>>合并merge，将溢写到本地磁盘的文件进行一次合并，合并完后每一个分区里也要排序

                              >>>最后形成一个大文件

            reduce端shuffle

                         >>到本地磁盘去拉取自己那部分数据

                         >>拉过来后放到内存，进行溢写操作

                         >>分组group

                              >>>将相同key的value放在一起 > 比较

                                      <hadoop, 1><hadoop, 1><hadoop, 1>

                                      <hadoop, list(1,1,1)>                                       reduce输入

           shuffle中的Combiner，可选项，是一种优化，减少了网络的传输，本地磁盘IO流的读写

           compress压缩，可配置项

14、网站基本指标分析

     （1）PV  page view，浏览量累计，每天，每周，每月。页面的浏览次数，衡量网站用户访问的网页数量，用户每打开一个页面就记录1次，多次打开同一页面则浏览量累计；

     （2）UV  unique visitor，独立访客数，userID，cookie可设置过期时间，session会话。1天内访问某站点的人数（以cookie为依据），1天内同一访客的多次访问只计为1个访客；

     （3）VV  visit view，访客的访问次数，session的统计。记录所有访客1天内访问了多少次您的网站，当访客完成浏览并关掉该网站的所有页面时便完成了一次访问，同一访客1天内可能有多次访问行为。

    （4）IP  独立IP数。指1天内使用不同IP地址的用户访问网站的数量，同一IP不管访问了几个页面，独立IP数均为1。查询本机IP：百度中输入IP。也可在http://ip.taobao.com中查询某个IP的所在位置。

    （5）跳出率。网站打开后什么都没有点击，马上退出比率。

    （6）案例PV统计

             需求：统计每天每个省份的PV值

             设计：provinceId   ->   key:id     value:1

                        map输出：<provinceId, list(1,1,1,1,1)>

                        reduce输出：<provinceId, 5>