hadoop，

hadoop，

1.1.Hadoop介绍
Hadoop：是一个由Apache基金会所开发的分布式系统基础架构 。
Hadoop适合应用于大数据存储和大数据分析的应用，适合于服务器几千台到几万台的集群运行，支持PB级的存储容量。
Hadoop典型应用有：搜索、日志处理、推荐系统、数据分析、视频图像分析、数据保存等。
Hadoop核心：HDFS: Hadoop Distributed File System 分布式文件系统
YARN: Yet Another Resource Negotiator 资源管理调度系统
Mapreduce：分布式运算框架
Hadoop的框架最核心的设计就是：HDFS和MapReduce。HDFS为海量的数据提供了存储，则MapReduce为海量的数据提供了计算

1.1.1.HDFS
hdfs是一种允许文件通过网络在多台主机上分享的文件系统，可让多机器上的多用户分享文件和存储空间。
通透性。让实际上是通过网络来访问文件的动作，由程序与用户看来，就像是访问本地的磁盘一般。
容错。即使系统中有某些节点脱机，整体来说系统仍然可以持续运作而不会有数据损失。
分布式文件管理系统很多，hdfs只是其中一种。适用于一次写入多次查询的情况，不支持并发写情况，小文件不合适。
安装hadoop
scp -r /usr/local/jdk mini2:/usr 远程复制
1.1.1.1HDFS结构
NameNode
DataNode
Secondary NameNode

NameNode
是整个文件系统的管理节点。
1.HDFS集群分为两大角色：NameNode、DataNode
2.NameNode负责管理整个文件系统的元数据
3.DataNode 负责管理用户的文件数据块
4.文件会按照固定的大小（blocksize）切成若干块后分布式存储在若干台datanode上
5.每一个文件块可以有多个副本，并存放在不同的datanode上
6.Datanode会定期向Namenode汇报自身所保存的文件block信息，而namenode则会负责保持文件的副本数量
7.HDFS的内部工作机制对客户端保持透明，客户端请求访问HDFS都是通过向namenode申请来进行

它维护着整个文件系统的文件目录树，文件/目录的元信息和每个文件对应的数据块列表。接收用户的操作请求。
NAMENODE职责：
负责客户端请求的响应
元数据的管理（查询，修改）
文件包括：
fsimage:元数据镜像文件。存储某一时段NameNode内存元数据信息。
edits:操作日志文件。
fstime:保存最近一次checkpoint的时间
以上这些文件是保存在linux的文件系统中。
HDFS写数据流程
概述：客户端要向HDFS写数据，首先要跟namenode通信以确认可以写文件并获得接收文件block的datanode，然后，客户端按顺序将文件逐个block传递给相应datanode，并由接收到block的datanode负责向其他datanode复制block的副本

详细步骤解析
1、根namenode通信请求上传文件，namenode检查目标文件是否已存在，父目录是否存在
2、namenode返回是否可以上传
3、client请求第一个 block该传输到哪些datanode服务器上
4、namenode返回3个datanode服务器ABC
5、client请求3台dn中的一台A上传数据（本质上是一个RPC调用，建立pipeline），A收到请求会继续调用B，然后B调用C，将整个pipeline建立完成，逐级返回客户端
6、client开始往A上传第一个block（先从磁盘读取数据放到一个本地内存缓存），以packet为单位，A收到一个packet就会传给B，B传给C；A每传一个packet会放入一个应答队列等待应答
7、当一个block传输完成之后，client再次请求namenode上传第二个block的服务器。

HDFS读数据流程
概述
客户端将要读取的文件路径发送给namenode，namenode获取文件的元信息（主要是block的存放位置信息）返回给客户端，客户端根据返回的信息找到相应datanode逐个获取文件的block并在客户端本地进行数据追加合并从而获得整个文件

1、跟namenode通信查询元数据，找到文件块所在的datanode服务器
2、挑选一台datanode（就近原则，然后随机）服务器，请求建立socket流
3、datanode开始发送数据（从磁盘里面读取数据放入流，以packet为单位来做校验）
4、客户端以packet为单位接收，先在本地缓存，然后写入目标文件

元数据存储机制
A、内存中有一份完整的元数据(内存meta data)
B、磁盘有一个“准完整”的元数据镜像（fsimage）文件(在namenode的工作目录中)
C、用于衔接内存metadata和持久化元数据镜像fsimage之间的操作日志（edits文件）
注：当客户端对hdfs中的文件进行新增或者修改操作，操作记录首先被记入edits日志文件中，当客户端操作成功后，相应的元数据会更新到内存meta.data中

元数据的checkpoint：
每隔一段时间，会由secondary namenode将namenode上积累的所有edits和一个最新的fsimage下载到本地，并加载到内存进行merge（这个过程称为checkpoint）

附带作用：namenode和secondary namenode的工作目录存储结构完全相同，所以，当namenode故障退出需要重新恢复时，可以从secondary namenode的工作目录中将fsimage拷贝到namenode的工作目录，以恢复namenode的元数据

dataNode
提供真实文件数据的存储服务。
Datanode工作职责：
存储管理用户的文件块数据
定期向namenode汇报自身所持有的block信息（通过心跳信息上报）

文件块（block）：最基本的存储单位。对于文件内容而言，一个文件的长度大小是size，那么从文件的０偏移开始，按照固定的大小，顺序对文件进行划分并编号，划分好的每一个块称一个Block。HDFS默认Block大小是128MB，以一个256MB文件，共有256/128=2个Block.
dfs.block.size
不同于普通文件系统的是，HDFS中，如果一个文件小于一个数据块的大小，并不占用整个数据块存储空间
Replication。多复本。默认是三个。

*****副本存放策略：

1、先在客户端所连接的datanode上存放一个副本
2、再在另一个机架上选择一个datanode存放第二个副本
3、最后在本机架上根据负载情况随机挑选一个datanode存放第三个副本

****副本数量的配置优先级
1、服务端hdfs-site.xml中可以配置
2、在客户端指定dfs.replication的值
客户端所指定的值优先级更高!!!

****定期向NameNode汇报自身所存储的block信息

1.1.1.2Rpc原理
即Remote Procdure Call，中文名：远程过程调用
RPC
同其他RPC框架一样，Hadoop RPC分为四个部分：
　　（1）序列化层：Clent与Server端通信传递的信息采用了Hadoop里提供的序列化类或自定义的Writable类型；
　　（2）函数调用层：Hadoop RPC通过动态代理以及java反射实现函数调用；
　　（3）网络传输层：Hadoop RPC采用了基于TCP/IP的socket机制；
（4）服务器端框架层：RPC Server利用java NIO以及采用了事件驱动的I/O模型，提高RPC Server的并发处理能力；
如NameNode和DataNode之间的通讯。

1.1.2.Mapreduce
基本思想：
1.将一个业务处理需求分成两个阶段来进行，map阶段，reduce阶段
2.将分布式计算中面临的公共的问题封装成框架来实现（jar包的分发，任务的启动，任务的容错，调度，中间结果的分组传递。）
3.应用开发人员只需要关心业务逻辑
4.Mapreduce只是分布式运算框架的一种实现，类似的框架还有很多，比如storm（流式计算），spark（内存迭代计算）

简单来说：将大的数据分析分成小块逐个分析，最后再将提取出来的数据汇总分析，最终获得我们想要的内容。

Map 和Reduce 是它的主要思想，来源于函数式编程语言，它的原理如下图所示：Map负责将数据打散，Reduce负责对数据进行聚集，用户只需要实现map 和reduce 两个接口，即可完成TB级数据的计算，常见的应用包括：日志分析和数据挖掘等数据分析应用。另外，还可用于科学数据计算，如圆周率PI 的计算等。Hadoop MapReduce的实现也采用了Master/Slave 结构。Master 叫做JobTracker，而Slave 叫做TaskTracker。用户提交的计算叫做Job，每一个Job会被划分成若干个Tasks。JobTracker负责Job 和Tasks 的调度，而TaskTracker负责执行Tasks。

1.1.1.3结构
一个完整的mapreduce程序在分布式运行时有三类实例进程：
1、MRAppMaster：负责整个程序的过程调度及状态协调
2、mapTask：负责map阶段的整个数据处理流程
3、ReduceTask：负责reduce阶段的整个数据处理流程

1.1.1.4流程解析
1、一个mr程序启动的时候，最先启动的是MRAppMaster，MRAppMaster启动后根据本次job的描述信息，计算出需要的maptask实例数量，然后向集群申请机器启动相应数量的maptask进程

2、maptask进程启动之后，根据给定的数据切片范围进行数据处理，主体流程为：
a)利用客户指定的inputformat来获取RecordReader读取数据，形成输入KV对
b)将输入KV对传递给客户定义的map()方法，做逻辑运算，并将map()方法输出的KV对收集到缓存
c)将缓存中的KV对按照K分区排序后不断溢写到磁盘文件

3、MRAppMaster监控到所有maptask进程任务完成之后，会根据客户指定的参数启动相应数量的reducetask进程，并告知reducetask进程要处理的数据范围（数据分区）

4、Reducetask进程启动之后，根据MRAppMaster告知的待处理数据所在位置，从若干台maptask运行所在机器上获取到若干个maptask输出结果文件，并在本地进行重新归并排序，然后按照相同key的KV为一个组，调用客户定义的reduce()方法进行逻辑运算，并收集运算输出的结果KV，然后调用客户指定的outputformat将结果数据输出到外部存储

1.1.1.5 MapTask并行度决定机制
maptask的并行度决定map阶段的任务处理并发度，进而影响到整个job的处理速度
那么，mapTask并行实例是否越多越好呢？其并行度又是如何决定呢？

mapTask并行度的决定机制
一个job的map阶段并行度由客户端在提交job时决定
而客户端对map阶段并行度的规划的基本逻辑为：
将待处理数据执行逻辑切片（即按照一个特定切片大小，将待处理数据划分成逻辑上的多个split），然后每一个split分配一个mapTask并行实例处理

这段逻辑及形成的切片规划描述文件，由FileInputFormat实现类的getSplits()方法完成，其过程如下图：

FileInputFormat切片机制
1、切片定义在InputFormat类中的getSplit()方法
2、FileInputFormat中默认的切片机制：
a)简单地按照文件的内容长度进行切片
b)切片大小，默认等于block大小
c)切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

FileInputFormat中切片的大小的参数配置
通过分析源码，在FileInputFormat中，计算切片大小的逻辑：Math.max(minSize, Math.min(maxSize, blockSize)); 切片主要由这几个值来运算决定
minsize：默认值：1
配置参数： mapreduce.input.fileinputformat.split.minsize
maxsize：默认值：Long.MAXValue
配置参数：mapreduce.input.fileinputformat.split.maxsize
blocksize
因此，默认情况下，切片大小=blocksize
maxsize（切片最大值）：
参数如果调得比blocksize小，则会让切片变小，而且就等于配置的这个参数的值
minsize （切片最小值）：
参数调的比blockSize大，则可以让切片变得比blocksize还大

选择并发数的影响因素：
1、运算节点的硬件配置
2、运算任务的类型：CPU密集型还是IO密集型
3、运算任务的数据量
map并行度的经验之谈
如果硬件配置为2*12core + 64G，恰当的map并行度是大约每个节点20-100个map，最好每个map的执行时间至少一分钟。
如果job的每个map或者 reduce task的运行时间都只有30-40秒钟，那么就减少该job的map或者reduce数，每一个task(map|reduce)的setup和加入到调度器中进行调度，这个中间的过程可能都要花费几秒钟，所以如果每个task都非常快就跑完了，就会在task的开始和结束的时候浪费太多的时间。
配置task的JVM重用可以改善该问题：
（mapred.job.reuse.jvm.num.tasks，默认是1，表示一个JVM上最多可以顺序执行的task
数目（属于同一个Job）是1。也就是说一个task启一个JVM）

如果input的文件非常的大，比如1TB，可以考虑将hdfs上的每个block size设大，比如设成256MB或者512MB

ReduceTask并行度的决定
reducetask的并行度同样影响整个job的执行并发度和执行效率，但与maptask的并发数由切片数决定不同，Reducetask数量的决定是可以直接手动设置：

//默认值是1，手动设置为4
job.setNumReduceTasks(4);

如果数据分布不均匀，就有可能在reduce阶段产生数据倾斜
注意： reducetask数量并不是任意设置，还要考虑业务逻辑需求，有些情况下，需要计算全局汇总结果，就只能有1个reducetask

尽量不要运行太多的reduce task。对大多数job来说，最好rduce的个数最多和集群中的reduce持平，或者比集群的 reduce slots小。这个对于小集群而言，尤其重要。

1.1.1.6程序运行模式
本地运行模式
（1）mapreduce程序是被提交给LocalJobRunner在本地以单进程的形式运行
（2）而处理的数据及输出结果可以在本地文件系统，也可以在hdfs上
（3）怎样实现本地运行？写一个程序，不要带集群的配置文件（本质是你的mr程序的conf中是否有mapreduce.framework.name=local以及yarn.resourcemanager.hostname参数）
（4）本地模式非常便于进行业务逻辑的debug，只要在eclipse中打断点即可

如果在windows下想运行本地模式来测试程序逻辑，需要在windows中配置环境变量：
％HADOOP_HOME％ = d:/hadoop-2.6.1
%PATH% = ％HADOOP_HOME％\bin
并且要将d:/hadoop-2.6.1的lib和bin目录替换成windows平台编译的版本

集群运行模式
（1）将mapreduce程序提交给yarn集群resourcemanager，分发到很多的节点上并发执行
（2）处理的数据和输出结果应该位于hdfs文件系统
（3）提交集群的实现步骤：
A、将程序打成JAR包，然后在集群的任意一个节点上用hadoop命令启动
$ hadoop jar wordcount.jar cn.itcast.bigdata.mrsimple.WordCountDriver inputpath outputpath
B、直接在linux的eclipse中运行main方法
（项目中要带参数：mapreduce.framework.name=yarn以及yarn的两个基本配置）
C、如果要在windows的eclipse中提交job给集群，则要修改YarnRunner类

mapreduce程序在集群中运行时的大体流程：

附：在windows平台上访问hadoop时改变自身身份标识的方法之二：

1.1.1.7. Combiner
（1）combiner是MR程序中Mapper和Reducer之外的一种组件
（2）combiner组件的父类就是Reducer
（3）combiner和reducer的区别在于运行的位置：
Combiner是在每一个maptask所在的节点运行
Reducer是接收全局所有Mapper的输出结果；
(4) combiner的意义就是对每一个maptask的输出进行局部汇总，以减小网络传输量
具体实现步骤：
1、自定义一个combiner继承Reducer，重写reduce方法
2、在job中设置： job.setCombinerClass(CustomCombiner.class)
(5) combiner能够应用的前提是不能影响最终的业务逻辑
而且，combiner的输出kv应该跟reducer的输入kv类型要对应起来

1.1.1.8 序列化
Java的序列化是一个重量级序列化框架（Serializable），一个对象被序列化后，会附带很多额外的信息（各种校验信息，header，继承体系。。。。），不便于在网络中高效传输；
所以，hadoop自己开发了一套序列化机制（Writable），精简，高效

如果需要将自定义的bean放在key中传输，则还需要实现comparable接口，因为mapreduce框中的shuffle过程一定会对key进行排序,此时，自定义的bean实现的接口应该是：
public class FlowBean implements WritableComparable

自定义的数据类型bean实现了Writable接口后，要实现其中的两个方法
public void write(DataOutput out) throws IOException ----序列化，将数据写入字节流
以及
public void readFields(DataInput in) throws IOException ----反序列化，从字节流中读出数据
注意：
写入数据和读出数据的顺序和类型要保持一致

1.1.1.9自定义分区和排序
自定义排序
hadoop的排序是在shuffle中完成的
排序的依据是map输出的key
要想实现自定义的排序，就要将需要排序的数据封装到key中传输，并且要将数据实现WritableComparable接口

自定义分区 partition
****每一个reduce task输出一个结果文件
----自定义一个类AreaPartitioner 继承 Partitioner 抽象类，实现其中的方法 int getPartition(K,V)
----在job的描述中设置使用自定义的partitioner
job.setPartitionerClass(AreaPartitioner.class)
----在job的描述中指定作业的reduce task并发数量，job.setNumReduceTasks(5)，数量要与partitioner中的分区数一致

1.1.1.10 shuffle机制
3.1.1 概述：
mapreduce中，map阶段处理的数据如何传递给reduce阶段，是mapreduce框架中最关键的一个流程，这个流程就叫shuffle；
shuffle: 洗牌、发牌——（核心机制：数据分区，排序，缓存）；
具体来说：就是将maptask输出的处理结果数据，分发给reducetask，并在分发的过程中，对数据按key进行了分区和排序；

Shuffle缓存流程：

shuffle是MR处理流程中的一个过程，它的每一个处理步骤是分散在各个map task和reduce task节点上完成的，整体来看，分为3个操作：
1、分区partition
2、Sort根据key排序
3. Combiner进行局部value的合并

详细流程
1、maptask收集map()方法输出的kv对，放到内存缓冲区中
2、从内存缓冲区不断溢出本地磁盘文件，可能会溢出多个文件
3、多个溢出文件会被合并成大的溢出文件
4、在溢出和合并过程中，都要进行分组和排序 （调用partitoner进行分组和针对key进行排序）
5、reducetask根据自己的分区号，去各个maptask机器上取相应的结果分区数据
6、reducetask会取到同一个分区的来自不同maptask的结果文件，再进行合并（归并排序）

7、合并成大文件后，shuffle的过程也就结束了，后面进入reducetask的逻辑运算过程（从文件中取出一个一个的键值对group，调用用户自定义的reduce()方法）

Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快 （从内存缓冲区不断溢出本地磁盘文件）
缓冲区的大小可以通过参数调整, 参数：io.sort.mb 默认100M

Partition（多少个reducer就有多少个partition）多节点的相同key合并，sort键值对哈希码排序
Combine 单节点上相同的key合并

在MapReduce计算框架中，主要用到了两种排序方法：快速排序和归并排序
问你有七八亿客户访问，有什么优化的方法计算pv

1.1.3.YARN
概述：Yarn是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台，而mapreduce等运算程序则相当于运行于操作系统之上的应用程序
YARN的重要概念
1、yarn并不清楚用户提交的程序的运行机制
2、yarn只提供运算资源的调度（用户程序向yarn申请资源，yarn就负责分配资源）
3、yarn中的主管角色叫ResourceManager
4、yarn中具体提供运算资源的角色叫NodeManager
5、这样一来，yarn其实就与运行的用户程序完全解耦，就意味着yarn上可以运行各种类型的分布式运算程序（mapreduce只是其中的一种），比如mapreduce、storm程序，spark程序，tez ……
6、所以，spark、storm等运算框架都可以整合在yarn上运行，只要他们各自的框架中有符合yarn规范的资源请求机制即可
7、Yarn就成为一个通用的资源调度平台，从此，企业中以前存在的各种运算集群都可以整合在一个物理集群上，提高资源利用率，方便数据共享

mapreduce程序的调度过程

Job的执行流程：
1.提交 - - RunJar的客户端跟ResourceManager协作完成job的提交
2.分配资源 - - resourcemanager和nodemanager协作完成job运行所需要的资源的分配
3.运行job ——mareduce框架中的主管进程MARppMaster负责整个job运行过程的协调控制

1.runjar向resourcemanager申请一个application，返回资源路径。
2.提交job运行需要的资源路径（放到那个路径）。
3.提交完毕后，申请运行MR

1.1.4.Hadoop的HA机制
前言：正式引入HA机制是从hadoop2.0开始，之前的版本中没有HA机制
1.1.1.11HA的运作机制
（1）hadoop-HA集群运作机制介绍
所谓HA，即高可用（7*24小时不中断服务）
实现高可用最关键的是消除单点故障
hadoop-ha严格来说应该分成各个组件的HA机制——HDFS的HA、YARN的HA

（2）HDFS的HA机制详解
通过双namenode消除单点故障
双namenode协调工作的要点：
A、元数据管理方式需要改变：
EDITS管理服务
内存中各自保存一份元数据
Edits日志只能有一份，只有Active状态的namenode节点可以做写操作
两个namenode都可以读取edits
共享的edits放在一个共享存储中管理（qjournal和NFS两个主流实现）
JournalNode服务器：运行的JournalNode进程非常轻量，可以部署在其他的服务器

active会把最近的操作记录写到本地一个edits文件，传到qj中，standy定期检查把最近的edits文件读取过来和fsiamge文件合并，成一个新的fsimage。合并完成之后会通知active来获取这个新的fsimage
Active获取到最新的之后重命名替换旧的fsimage文件

B、需要一个状态管理功能模块

实现了一个zkfailover，常驻在每一个namenode所在的节点
每一个zkfailover负责监控自己所在namenode节点，利用zk进行状态标识
当需要进行状态切换时，由zkfailover来负责切换
切换时需要防止brain split现象的发生

Zookeeper：
是一个高可用的分布式数据管理与系统协调框架。是 Hadoop 的分布式协调服务
Hadoop2.0,使用Zookeeper的事件处理确保整个集群只有一个活跃的NameNode,存储配置信息等.
HBase,使用Zookeeper的事件处理确保整个集群只有一个HMaster,察觉HRegionServer联机和宕机,存储访问控制列表等.
统一命名服务、配置管理、集群管理、共享锁、队列管理

用来维护和监控存储的数据的状态变化通过监控这些数据状态的变化从而达到基于数据的集群管理
文件系统+通知机制