3.4.MR的编程
编程分析：
map任务处理： 读取输入文件内容，解析成key、value对。对输入文件的每一行，解析成key、value对。每一个键值对调用一次map函数。 写自己的逻辑，对输入的key、value处理，转换成新的key、value输出。 对输出的key、value进行分区。 对不同分区的数据，按照key进行排序、分组。相同key的value放到一个集合中。 (可选)分组后的数据进行归约。 reduce任务处理： 对多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点。 对多个map任务的输出进行合并、排序。写reduce函数自己的逻辑，对输入的key、value处理，转换成新的key、value输出。 把reduce的输出保存到文件中。
编程代码：
public class WordCount { //继承mapper接口，设置map的输入类型为<Object,Text> //输出类型为<Text,IntWritable> public static class Map extends Mapper<Object,Text,Text,IntWritable>{ //one表示单词出现一次 private static IntWritable one = new IntWritable(1); //word存储切下的单词 private Text word = new Text(); public void map(Object key,Text value,Context context) throws IOException,InterruptedException{ //对输入的行切词 StringTokenizer st = new StringTokenizer(value.toString()); while(st.hasMoreTokens()){ word.set(st.nextToken());//切下的单词存入word context.write(word, one); } } } //继承reducer接口，设置reduce的输入类型<Text,IntWritable> //输出类型为<Text,IntWritable> public static class Reduce extends Reducer<Text,IntWritable,Text,IntWritable>{ //result记录单词的频数 private static IntWritable result = new IntWritable(); public void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException,InterruptedException{ int sum = 0; //对获取的<key,value-list>计算value的和 for(IntWritable val:values){ sum += val.get(); } //将频数设置到result result.set(sum); //收集结果 context.write(key, result); } } /** * @param args */ public static void main(String[] args) throws Exception{ // TODO Auto-generated method stub Configuration conf = new Configuration(); conf.set("mapred.job.tracker", "localhost:9001"); args = new String[]{"hdfs://localhost:9000/user/hadoop/input/count_in","hdfs://localhost:9000/user/hadoop/output/count_out"}; //检查运行命令 String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs(); if(otherArgs.length != 2){ System.err.println("Usage WordCount <int> <out>"); System.exit(2); } //配置作业名 Job job = new Job(conf,"word count"); //配置作业各个类 job.setJarByClass(WordCount.class); job.setMapperClass(Map.class); job.setCombinerClass(Reduce.class); job.setReducerClass(Reduce.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
4.Yarn专题
4.1.yarn简介
YARN 是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台。Spark、Storm 等运算框架都可以整合在 YARN 上运行，只要他们各自的框架中有符合 YARN 规范的资源请求机制即可
设计思想
YARN 的基本思想是将资源管理和作业调度/监控的功能拆分为单独的守护进程。这个想法是有一个全局 ResourceManager ( RM ) 和每个应用程序 ApplicationMaster ( AM )。应用程序是单个作业或作业的 DAG。
ResourceManager 和 NodeManager 构成了数据计算框架。ResourceManager 是在系统中的所有应用程序之间仲裁资源的最终权威。NodeManager 是每台机器的框架代理，负责容器、监控它们的资源使用情况（cpu、内存、磁盘、网络）并将其报告给 ResourceManager/Scheduler。
每个应用程序的 ApplicationMaster 实际上是一个特定于框架的库，其任务是协商来自 ResourceManager 的资源并与 NodeManager 一起执行和监视任务。
4.2.yarn架构与服务角色
架构图如下：
2.1ResourceManager
RM(ResourceManager)是 YARN 集群主控节点，负责协调和管理整个集群（所有 NodeManager）的资源。ResourceManager 会为每一个 Application 启动一个 AppMaster，并且 AppMaster 分散在各个 NodeManager 节点。
ResourceManager 的职责：
1、处理客户端请求，比如程序提交
2、启动或监控 MRAppMaster
3、监控 NodeManager健康状态
4、资源的分配与调度
ResourceManager 有两个主要组件：Scheduler 和 ApplicationsManager。
Scheduler 调度器负责根据熟悉的容量、队列等约束将资源分配给各种正在运行的应用程序。调度器是纯粹的调度器，因为它不
2.2NodeManager
NodeManager 是 YARN 集群当中真正资源的提供者，是真正执行应用程序的容器的提供者，监控应用程序的资源使用情况（CPU，内存，硬盘，网络），并通过心跳向集群资源调度器ResourceManager 进行汇报以更新自己的健康状态。同时其也会监督 Container 的生命周期管理，监控每个 Container 的资源使用（内存、CPU 等）情况，追踪节点健康状况，管理日志和不同应用程序用到的附属服务（auxiliary service）。
2.3AppMaster
AppMaster对应一个应用程序，职责是：向YARN资源调度器申请执行任务的资源容器，运行任务，监控整个任务的执行，跟踪整个任务的状态，处理任务失败以异常情况
2.4Container容器
Container 容器是一个抽象出来的逻辑资源单位。容器是由 ResourceManager Scheduler 服务动态分配的资源构成，它包括了一个节点上的一定量 CPU，内存，磁盘，网络等信息，容器的大小是可以动态调整的,其最大大小就是一个nodemanger的资源，一个container只能在一个容器上。
4.3.Scheduler调度器
调度器根据应用程序的资源需求进行资源分配，不参与应用程序具体的执行和监控等工作，调度器会将总资源分为不同的队列供共享资源进行分配。资源分配的单位就是 Container，YARN 本身为我们提供了多种直接可用的调度器，比如 FIFO队列调度器，Fair 公平调度器和 Capacity 容量调度器等，可以在conf/yarn-site.xml 中配置调度器。
1.CapacityScheduler容量调度器
其中心思想是 Hadoop 集群中的可用资源在多个组织之间共享，这些组织根据其计算需求共同资助集群，为了保证集群总资源的安全和稳定性，CapacityScheduler提供了一组严格的限制，以确保单个应用程序或用户或队列不能在集群中消耗的资源的量不成比例。此外，CapacityScheduler对来自单个用户和队列的初始化和挂起的应用程序提供限制，以确保集群的公平性和稳定性。简单说就是当目前使用队列资源不够时，动态分配其他队列资源进行使用。参数配置指定调度器：yarn.resourcemanager.scheduler.class=指定挂起和运行最大程序个数：默认是10000：maximum-applications集群中程序最大可占用的运行资源：maximum-am-resource-percent默认是1-10%=90%的最大资源使用率，留10%给系统指定某用户的最大运行程序数：yarn.scheduler.capacity.user..max-parallel-apps 可以设置单个队列的资源占比，运行app任务数等。
2.Fair 公平调度器公平调度
是一种将资源分配给应用程序的方法，以便所有应用程序在一段时间内平均获得相等的资源份额。默认情况下，Fair Scheduler 仅基于内存进行调度公平性决策。简单来说就是对于两个程序抢夺同一资源时进行均分资源，保证都能得到一半。
3.FIFO Scheduler(先进先出调度器)
FIFO Scheduler把应用按提交的顺序排成一个队列，这是一个先进先出队列，在进行资源分配的时候，先给队列中最头上的应用进行分配资源，待最头上的应用需求满足后再给下一个分配，以此类推。FIFO Scheduler是最简单也是最容易理解的调度器，也不需要任何配置，但它并不适用于共享集群。大的应用可能会占用所有集群资源，这就导致其它应用被阻塞。在共享集群中，更适合采用Capacity Scheduler或Fair Scheduler，这两个调度器都允许大任务和小任务在提交的同时获得一定的系统资源。
4.4.YARN 作业执行流程
利用yarn进行资源调度的程序，基本流程都和下面一致，包括spark,MR等数据处理组件
1.client向yarn提交job，首先找ResourceManager分配资源，
2.ResourceManager开启一个Container,在Container中运行一个Application manager
3.Application manager找一台nodemanager启动Application master，计算任务所需的计算
4.Application master向Application manager（Yarn）申请运行任务所需的资源
5.Resource scheduler将资源封装发给Application master
6.Application master将获取到的资源分配给各个nodemanager
7.各个nodemanager得到任务和资源开始执行map task
8.map task执行结束后，开始执行reduce task
9.map task和 reduce task将执行结果反馈给Application master
10.Application master将任务执行的结果反馈pplication manager
5.HIVE专题
5.1HIVE简介
Hive是一个数据仓库基础工具在Hadoop中用来处理结构化数据。它架构在Hadoop之上，总归为大数据，并使得查询和分析方便。并提供简单的sql查询功能，可以将sql语句转换为MapReduce任务进行运行。
最初，Hive是由Facebook开发，后来由Apache软件基金会开发，并作为进一步将它作为名义下Apache Hive为一个开源项目。Hive 没有专门的数据格式。 Hive 可以很好的工作在 Thrift 之上，控制分隔符，也允许用户指定数据格式。Hive不适用于在线事务处理。 它最适用于传统的数据仓库任务。
Hive 构建在基于静态批处理的Hadoop 之上，Hadoop 通常都有较高的延迟并且在作业提交和调度的时候需要大量的开销。因此，Hive 并不能够在大规模数据集上实现低延迟快速的查询，例如，Hive 在几百MB 的数据集上执行查询一般有分钟级的时间延迟。因此，
Hive 并不适合那些需要低延迟的应用，例如，联机事务处理（OLTP）。Hive 查询操作过程严格遵守Hadoop MapReduce 的作业执行模型，Hive 将用户的HiveQL 语句通过解释器转换为MapReduce 作业提交到Hadoop 集群上，Hadoop 监控作业执行过程，然后返回作业执行结果给用户。Hive 并非为联机事务处理而设计，Hive 并不提供实时的查询和基于行级的数据更新操作。Hive 的最佳使用场合是大数据集的批处理作业，例如，网络日志分析。
直接使用hadoop所面临的问题：
人员学习成本太高
项目周期要求太短
MapReduce实现复杂查询逻辑开发难度太大
使用HIVE可解决的问题：
操作接口采用类SQL语法，提供快速开发的能力。
避免了去写MapReduce，减少开发人员的学习成本。
扩展功能很方便。
HIVE架构
5.2HIVE如何执行一条HQL语句
概述：
第一步：输入一条HQL查询语句（select * from tab） 第二步：解析器对这条Hql语句进行语法分析。 第三步：编译器对这条Hql语句生成HQL的执行计划。 第四步：优化器生成最佳的Hql的执行计划。 第五步：执行这条最佳Hql语句。
比如如下HQL语句：
select dept, sum(salary) from emp group by dept;
考虑下这个东西你自己写会怎么计算：你有一张表在hdfs上，这个表是一张员工表，有几个字段：
id:integer，name:varchar，dept:varchar，memo: string，salary：Integer 1，张三，总裁办，暂缺，1000 2，李四，总裁办，暂缺，1000 3，王五，外联部，无，2000
上面数据采样的第一列是ID，后面是名字，接着是DEPT（部门），最后是备注和工资。然后SQL是用来统计每个部门工资支出的。GROUP BY DEPT就是按照DEPT分组，把相同的放在一起，然后各自算出工资总额。单机做的话就是先按照部门排个序，然后一个大循环单独统计每个部门的总额。最后一个部门输出一条结果。
如果是Hive呢，大致上也类似，只是搬到分布式环境下了。
Hive基本上会遵循AST->逻辑执行计划->优化的逻辑执行计划->物理执行计划->打包提交->分布式执行，这样的顺序来开展计算。
现在详细的说明开始，从SQL字符串输入开始。
SQL输入是个字符串，Hive需要先把字符串分解成自己能明白的结构。Hive的解析不是徒手撸的，用的是ANTLR，一个解析器生成器，可以用类似BNF的范式定义语法然后产生对应的解析代码。BNF是大学编译原理必学的东西，另外Hive的语法定义是一种后缀是.g的特殊文件，可以自己去Hive的代码里搜一下。
ANTLR生成的代码会返回给你ASTNode。AST是抽象语法树的简写。比如上面的SELECT，会转化为一个以TOK_SELECT标记为根节点的树，树的叶子节点需要包含Projection List子树（dept，count(*)），FROM子树，Filter子树（上面的例子空缺了）等等。也就是说，AST会把一个长字符串转化成树结构，树本身的结构设计取决于你的语法定义，ANTLR会按照你的定义把树排列好，Hive自身的代码而言，也要根据树结构定义去遍历整个树，把需要的信息抽取出来。AST的好处是，你不再纠结于Token的解析和排列问题，你只需要在一个固定结构的树上抽取信息，比如SELECT根节点以下你必然能找到SELECT_EXPR子节点（就是Projection部分的信息），很适合做这件事情。
遍历完整个AST，Hive把它关心的信息分类组织排列到一个结构中，但是还没有进行元信息绑定和检查整理，而这个绑定整理的过程叫Semantics Analyze（语义分析）。对应上面的例子，Hive会分别抽取：
Projection列表，上述例子就是DEPT，sum(salary)。这里拿到的信息表示，用户希望结果是什么表达式； FROM列表，表示数据源； GROUP BY，表示如何分组。
然后要做的是元数据绑定。这里首先需要从Hive元数据库中查询到相关的元信息。对上面的查询，Hive知道了用户希望从emp表中查询数据，那么Hive调用MetastoreClient接口，从Metadata Service中抽取了emp表的元信息，所谓元信息最基本地包含了表的schema，比如id是Integer类型，dept是string类型，这些信息都会注入本次之行Hive的符号解析空间，同时被注入的符号还有内建函数（比如我们用的sum）和UDF等等。然后Hive对Projection列表中的表达式进行解析，首先是dept，Hive会去搜索刚才提到的符号解析空间，找到了DEPT代表源表中的一个字段，类型是String。因为这里就是一个简单的字段引用，因此不涉及类型检查之类的；但是如果你写DEPT+1，这里就会有类型检查和隐式转换等，比如是强类型的设计，看到string＋integer的组合就会抛出异常结束执行，而弱类型的系统会偷偷插入一个CAST，把1转换成字符串“1”。第二列是sum(salary)，这里Hive看到一个函数调用，于是它搜索了函数表，找到了sum函数，并看到它是一个聚合函数（sum不是简单的一行数据能独立计算的函数，需要整个组一起算），这里就先标记起来，后面还有相关的语义检查。而继续遍历sum的子树又发现了一个salary列引用表达式，一样做一次解析找到salary列定义是Integer。
最后是Group By检查，根据SQL语义，出现在聚合函数外的字段引用必须出现在Group By中，于是Hive开始检查sum(SALARY)之外的列引用，发现了DEPT，然后遍历Group By的列表进行匹配，发现所有非聚合列都已经在GROUP BY中定义了，于是Hive很满意，继续执行下去了。
到这里为止Hive得到了一个语义正确的SQL查询信息结构。接下去需要的是，产生逻辑执行计划。
逻辑执行计划可以简单地认为就是，按照顺序在单机上跑是能跑出结果的一个计算计划。
按照最初我们分析的算法，首先你需要扫描整个表，因此Hive先产生一个TableScanOperator；接下去，需要做的是抽取DEPT字段和SALARY（sum的输入参数需要比sum先计算完），因此这两个操作在一起生成一个ProjectionOperator；然后是AggregationOperator，就是sum本身的计算，当然这个计算还需要携带DEPT的部分；最后是输出，就是SinkOperator，需要把结果写到HDFS上。
接下去Hive会对执行计划进行优化，最常见的优化可能是PartitionPrune，比如你在Hive中定义了分区表，那么如果有Where条件中出现了分区字段，比如WHERE date ＝ ‘2016-08-25’，而且分区就是date，那么我需要在TableScanOperator中加入分区信息，指定Scan的时候只扫描2016-8-25的信息；或者如果你有子查询，Hive会将内层的查询条件推送到外层，这样需要计算的数据就会减少；再或者如果底层存储支持列裁剪，那么刚才那张表其实我只用到了DEPT和SALARY两个字段，ID和名字以及备注我都不关心，那么Scan的时候可以少读一些信息。更复杂的还有比如CBO相关优化，可以交换Join顺序，让多次Join产生的中间数据尽可能小，或者选择不同的JOIN策略等等。
优化完毕之后，Hive接着生成物理执行计划。所谓物理执行计划才是真正映射底层计算引擎的计算策略。根据底层引擎不同（现在Hive支持Spark，Tez和MapReduce），Hive会生成不同的物理对应。拿MapReduce来说，刚才的SQL到这个阶段，需要拆解成Mapper和Reducer不同的步骤。TableScan，Projection计算以及Sum的前半段都是需要塞到Mapper做的。比较特殊的是Sum，因为Sum在MapReduce环境下需要在Mapper中计算Sum括号内的表达式（这里就是简单的提取SALARY），并且在每个Mapper中进行本地累加（Combiner中根据分组进行本地加和），然后分发到Reducer进行最终累加。因此AggregationOperator在物理执行计划产生的时候会拆解成两部分，一部分是PartialAggregation对应Mapper端的Sum，一部分是FinalAggregation对应Reducer端的最终Sum。这样我们的执行计划就变成：
