北京大学hadoop考试复习重点

1.Hdfs和Mapreduce核心 2.扩展存储Hbase，MDB 3.搭建平台的思路（平台堆栈） 4.分布式协调 zookeeper 5.分布式概念，系统，框架（1道题） 6.cap理论

Hadoop运行分三种方式： 单机、 伪集群、 集群。¨  Hadoop部署好后， 默认就是单机模式。 只在有任务的时候才会启动Hadoop进程， 并且只有一个进程， 该进程将完成所有计算任务。¨  必须要修改配置文件才能启动伪集群和集群模式。 这两种模式将按正常方式启动所有进程，也是常用的方式。 如果要启动伪集群模式， 只要把这些配置为本机回环地址（localhost） 即可。如果要启动集群模式， 只要把这些配置为集群中对应的主机名即可。 Hadoop启动后会在各节点启动WEB-UI， 管理员可以通过浏览器访问指定端口来查看集群或各节点的基本信息。NameNode的默认监听端口是50070， DataNode的默认监听端口是50075，Yarn资源管理器的默认监听端口是8088。

“一个分布式系统是若干个独立的计算机的集合， 但是对该系统的用户来说， 系统就像一台计算机一样。” 硬件方面： 各个计算机都是自治的 •  软件方面： 用户将整个系统看作是一台计算机。

分布式系统的特点 经济性： 高的性能/价格比 性能： 能产生单个大型主机不能达到的绝对性能 应用的固有的分布性, 例如： •  CSCW •  工作流 4、 可靠性高 •  多工系统的容错能力 5、 可扩充性强 •  系统演进能力（evolution）

HDFS主从结构体系 ØNameNode： 主控制服务器， 负责维护文件系统的命名空间（ Namespace） 并协调客户端对文件的访问， 记录命名空间内的任何改动或命名空间本身的属性改动 ØDataNode： 负责它们所在的物理节点上的存储管理

1． 冗余备份 每个文件存储成一系列数据块（ Block） ， 默认块大小为64MB（可配置） 。 为了容错，文件的所有数据块都会有副本（ 副本数量即复制因子， 可配置） 2． 副本存放 采用机架感知（ Rack-aware）的策略来改进数据的可靠性、可用性和网络带宽的利用率 3.心跳检测 NameNode周期性地从集群中的每个DataNode接受心跳包和块报告， 收到心跳包说明该DataNode工作正常 4.安全模式 系统启动时， NameNode会进入一个安全模式。 此时不会出现数据块的写操作 5.数据完整性检测 HDFS客户端软件实现了对HDFS文件内容的校验和（ Checksum）检查

6.空间回收 文件被用户或应用程序删除时， 先把它移动到/trash目录里；只要还在这个目录里， 文件就可以被迅速恢复 7.元数据磁盘失效 NameNode可以配置为支持维护映像文件和事务日志的多个副本，任何对映像文件或事务日志的修改， 都将同步到它们的副本上 8.快照 快照支持存储某个时间的数据复制， 当HDFS数据损坏时， 可以回滚到过去一个已知正确的时间点。 HDFS目前还不支持快照功能

对于任何大容量的分布式存储系统而言， 文件 压缩都是必须的， 文件压缩带来了两个好处： 减少了文件所需的存储空间： 加快了文件在网络上或磁盘间的传输速度。

Zookeeper（又称分布式锁） 是由开源组织Apache开发的一个的高效、 可靠 的分布式协调服务。

HBase 是什么？

是一个高可靠性、 高性能、 列存储、 可伸缩、 实时读写的分布式数据库系统。

应用场景1－统一命名服务

分布式应用中， 通常需要有一套完整的命名规则，既能够产生唯一的名称又便于人识别和记住， 通常情况下用树形的名称结构是一个理想的选择， 树形的名称结构是一个有层次的目录结构， 既对人友好又不会重复。 »  Name Service 是 Zookeeper 内置的功能，只要调⽤Zookeeper 的 API 就能实现

应用场景2－配置管理

将配置信息保存在 Zookeeper 的某个目录节点中，然后将所有需要修改的应用机器监控配置信息的状态， 一旦配置信息发生变化， 每台应用机器就会收到 Zookeeper 的通知， 然后从 Zookeeper 获取新的配置信息应用到系统中。

应用场景3－集群管理

Zookeeper 能够很容易的实现集群管理的功能， 如有多台 Server 组成一个服务集群， 那么必须要一个“总管” 知道当前集群中每台机器的服务状态，一旦有机器不能提供服务， 集群中其它集群必须知道， 从而做出调整重新分配服务策略。 同样当增加集群的服务能力时， 就会增加一台或多台 Server， 同样也必须让“总管” 知道。 »  Zookeeper 不仅能够维护当前的集群中机器的服务状态， 而且能够选出一个“总管” ， 让这个总管来管理集群， 这就是 Zookeeper 的另一个功能 LeaderElection。 规定编号最小的为master,所以当我们对SERVERS节点做监控的时候， 得到服务器列表， 只要所有集群机器逻辑认为最小编号节点为master， 那么master就被选出， 而这个master宕机的时候， 相应的znode会消失， 然后新的服务器列表就被推送到客户端， 然后每个节点逻辑认为最小编号节点为master， 这样就做到动态master选举。

应用场景4－共享锁

共享锁在同一个进程中很容易实现， 但是在跨进程或者在不同 Server 之间就不好实现了。 Zookeeper却很容易实现这个功能， 实现方式也是需要获得锁的Server 创建一个 EPHEMERAL_SEQUENTIAL 目录节点， 然后调用 getChildren方法获取当前的目录节点列表中最小的目录节点是不是就是自己创建的目录节点， 如果正是自己创建的， 那么它就获得了这个锁， 如果不是那么它就调用 exists(String path,boolean watch) 方法并监控 Zookeeper 上目录节点列表的变化， 一直到自己创建的节点是列表中最小编号的目录节点， 从而获得锁， 释放锁很简单， 只要删除前面它自己所创建的目录节点就行了。

应用场景5－队列管理

»  Zookeeper 可以处理两种类型的队列： 当一个队列的成员都聚齐时， 这个队列才可用， 否则一直等待所有成员到达， 这种是同步队列； 队列按照 FIFO方式进行入队和出队操作， 例如实现生产者和消费者模型。

.1 SPARK是什么 Spark 是基于内存计算的大数据并行计算框架。Spark 基于内存计算， 提高了在大数据环境下数据处理的实时性， 同时保证了高容错性和高可伸缩性， 允许用户将Spark 部署在大量廉价硬件之上， 形成集群。 Spark 于2009 年诞生于加州大学伯克利分校AMPLab。 目前， 已经成为Apache 软件基金会旗下的顶级开源项目。 下面是Spark 的发展历程。

CAP理论

1、简要描述如何安装配置一个apache开源版hadoop，描述即可，列出步骤更好

    --解压hadoop包，到指定安装文件夹

    --配置linux基本网络环境、jdk环境、防火墙环境

    --修改主机名，方便后面UI的访问

    --修改hadoop/etc/hadoop/conf下的配置文件，根据部署的模式和需要进行配置

    --格式化namenode，对数据缓存的的路径进行格式化

    --启动hadoop进程

2、请列出正常工作的hadoop集群中hadoop都需要启动哪些进程，他们的作用分别是什么？

    --namenode =>HDFS的守护进程，负责维护整个文件系统，存储着整个文件系统的元数据信息，有image+edit log namenode不会持久化存储这些数据，而是在启动时重建这些数据。

    --datanode  =>是具体文件系统的工作节点，当我们需要某个数据，namenode告诉我们去哪里找，就直接和那个DataNode对应的服务器的后台进程进行通信，由DataNode进行数据的检索，然后进行具体的读/写操作

    --secondarynamenode  =>一个冗余的守护进程，相当于一个namenode的元数据的备份机制，定期的更新，和namenode进行通信，将namenode上的image和edits进行合并，可以作为namenode的备份使用

    --resourcemanager =>是yarn平台的守护进程，负责所有资源的分配与调度，client的请求由此负责，监控nodemanager

    --nodemanager  => 是单个节点的资源管理，执行来自resourcemanager的具体任务和命令

3、启动hadoop报如下错误，该如何解决？

--1.error  org.apache.hadoop.hdfs.server.namenode.NameNode 

    --找不到主类，应该是配置文件的hadoop的安装位置配置错误，对hadoop-env.sh文件进行检查修改

--2.org.apache.hadoop.hdfs.server.common.inconsistentFSStateException

    --这个是存储目录不存在，或者被删除，对namenode进行格式化，或重新格式化，对tmp.dir进行自己的设置

--3.Directory /tmp/hadoop-root/dfs/name is in an inconsistent 

    --这个和上面一样的，重新设置core-site.xml中hadoop.tmp.dir的值，对namenode进行格式化，

--4.state storage direction does not exist or is not accessible?

    --之前是默认的tmp目录，每次重启都会清除这个数据，所以找不到整个文件系统的信息，重新设置core-site.xml中hadoop.tmp.dir的值，对namenode进行格式化，

4、请列出你所知道的hadoop调度器，并简要说明其工作方法？

--1.先进先出调度器（FIFO）

--Hadoop 中默认的调度器，也是一种批处理调度器。它先按照作业的优先级高低，再按照到达时间的先后选择被执行的作业

–2.容量调度器（Capacity Scheduler)

–支持多个队列，每个队列可配置一定的资源量，每个队列采用FIFO调度策略，为了防止同一个用户的作业独占队列中的资源，该调度器会对同一用户提交的作业所占资源量进行限定。调度时，首先按以下策略选择一个合适队列：计算每个队列中正在运行的任务数与其应该分得的计算资源之间的比值，选择一个该比值最小的队列；然后按以下策略选择该队列中一个作业：按照作业优先级和提交时间顺序选择，同时考虑用户资源量限制和内存限制

–3.公平调度器（Fair Scheduler）

–公平调度是一种赋予作业（job）资源的方法，它的目的是让所有的作业随着时间的推移，都能平均的获取等同的共享资源。所有的 job 具有相同的资源,当单独一个作业在运行时，它将使用整个集群。当有其它作业被提交上来时，系统会将任务（task）空闲资源（container）赋给这些新的作业，以使得每一个作业都大概获取到等量的CPU时间。与Hadoop默认调度器维护一个作业队列不同，这个特性让小作业在合理的时间内完成的同时又不"饿"到消耗较长时间的大作业。公平调度可以和作业优先权搭配使用——优先权像权重一样用作为决定每个作业所能获取的整体计算时间的比例。同计算能力调度器类似，支持多队列多用户，每个队列中的资源量可以配置， 同一队列中的作业公平共享队列中所有资源。

5、当前日志采样格式为如下，请编写MapReduce计算第四列每个元素出现的个数

a,b,c,d

a,s,d,f

d,f,g,c 就如此格式，

代码如下，比wordcount还要简单一点，代码差不多的

package make.hadoop.com.four_column;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner;

public class four_column extends Configured implements Tool { // 1、自己的map类 // 2、继承mapper类，<LongWritable, Text, Text, // IntWritable>输入的key,输入的value，输出的key,输出的value public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private IntWritable MapOutputkey = new IntWritable(1); private Text MapOutputValue = new Text();

	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {

		String strs = value.toString();
		// 分割数据
		String str_four = strs.split(",")[3];

		MapOutputValue.set(str_four);
		System.out.println(str_four);
		context.write(MapOutputValue, MapOutputkey);

	}
}
// 2、自己的reduce类，这里的输入就是map方法的输出
public static class MyReduce extends
		Reducer<Text, IntWritable, Text, IntWritable> {

	IntWritable countvalue = new IntWritable(1);

	@Override
	// map类的map方法的数据输入到reduce类的group方法中，得到<text,it(1,1)>,再将这个数据输入到reduce方法中
	protected void reduce(Text inputkey, Iterable<IntWritable> inputvalue,
			Context context) throws IOException, InterruptedException {

		int sum = 0;

		for (IntWritable i : inputvalue) {
			System.out.println(i.get());
			sum = sum + i.get();
		}
		// System.out.println("key: "+inputkey + "...."+sum);
		countvalue.set(sum);
		context.write(inputkey, countvalue);
	}
}
// 3运行类，run方法，在测试的时候使用main函数，调用这个类的run方法来运行

/**
 * param args 参数是接受main方得到的参数，在run中使用
 */
public int run(String[] args) throws Exception {

	Configuration conf = new Configuration();

	Job job = Job.getInstance(this.getConf(), "four_column");

	// set mainclass
	job.setJarByClass(four_column.class);

	// set mapper
	job.setMapperClass(MyMapper.class);
	job.setMapOutputKeyClass(Text.class);
	job.setMapOutputValueClass(IntWritable.class);

	// set reducer
	job.setReducerClass(MyReduce.class);
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(IntWritable.class);

	// set path
	Path inpath = new Path(args[0]);
	FileInputFormat.setInputPaths(job, inpath);
	Path outpath = new Path(args[1]);
	FileOutputFormat.setOutputPath(job, outpath);
	FileSystem fs = FileSystem.get(conf);
	// 存在路径就删除
	if (fs.exists(outpath)) {
		fs.delete(outpath, true);
	}
	job.setNumReduceTasks(1);

	boolean status = job.waitForCompletion(true);

	if (!status) {
		System.err.println("the job is error!!");
	}

	return status ? 0 : 1;

}
public static void main(String[] args) throws IOException,
		ClassNotFoundException, InterruptedException {

	Configuration conf = new Configuration();

	int atatus;
	try {
		atatus = ToolRunner.run(conf, new four_column(), args);
		System.exit(atatus);
	} catch (Exception e) {
		e.printStackTrace();
	}

}

}

6、hive有哪些方式保存元数据，各有哪些特点？

--1.内嵌Derby数据库存储

    --这个是hive默认自带的内嵌数据库，用来储存元数据，但这个在配置了hiveserver2和metastore服务后，不支持多个用户同时登录，不方便对数据库的安全访问

--2.multi user mode

    --在自己本地配一个，mysql的数据库用作，hive的元数据的存储数据库，这个需要要自己本地搭建一个mysql数据库，通过配置文件创建一个，hive自己的元数据库，也是我们学习一般会用的方式，配置一般如下

jasbdaksbdaskbdoajsbdasbu javax.jdo.option.ConnectionURL jdbc:mysql://hostname:3306/hive?createDatabaseIfNotExist=true JDBC connect string for a JDBC metastore hive.metastore.uris thrift://hostname:9083 IP address (or fully-qualified domain name) and port of the metastore host javax.jdo.option.ConnectionDriverName com.mysql.jdbc.Driver Driver class name for a JDBC metastore javax.jdo.option.ConnectionUserName xxxx username to use against metastore database javax.jdo.option.ConnectionPassword xxxx password to use against metastore database --3.remote server mode

   --一种在远端配置数据库服务的方式，这个需要配置metastore服务，通过客户端的metastore服务访问服务器上的元数据库达到访问数据的目的

7、 请简述hadoop怎么样实现二级排序？

    --在MapReduce中本身就会对我们key进行排序，所以我们要对value进行排序，主要思想为将key和部分value拼接成一个组合key（实现WritableComparable接口或者调用 setSortComparatorClass函数），这样reduce获取的结果便是先按key排序，后按value排序的结果，在这个方法中，用户需 要自己实现Paritioner，继承Partitioner<>,以便只按照key进行数据划分。Hadoop显式的支持二次排序，在Configuration类中有个 setGroupingComparatorClass()方法，可用于设置排序group的key值。

8、用非递归方法实现二分查找

    --代码如下，二分查找只适用于有序数列，对其进行查找，效率非常高，不适用于无序数列

public static int binSearch(int srcArray[], int key) { int mid; int start = 0; int end = srcArray.length - 1; while (start <= end) { mid = (end - start) / 2 + start; if (key < srcArray[mid]) { end = mid - 1; } else if (key > srcArray[mid]) { start = mid + 1; } else { return mid; } } return -1; } 递归的二分查找

public static int binSearch_di(int srcArray[], int start, int end, int key) { int mid = (end - start) / 2 + start; if (srcArray[mid] == key) { return mid; } if (start >= end) { return -1; } else if (key > srcArray[mid]) { return binSearch_di(srcArray, mid + 1, end, key); } else if (key < srcArray[mid]) { return binSearch_di(srcArray, start, mid - 1, key); } return -1; }

9、请简述mapreduce中，combiner，partition作用？

    --在MapReduce整个过程中，combiner是可有可无的，需要是自己的情况而定，如果只是单纯的对map输出的key-value进行一个统计，则不需要进行combiner，combiner相当于提前做了一个reduce的工作，减轻了reduce端的压力，

Combiner只应该适用于那种Reduce的输入（key：value与输出（key：value）类型完全一致，且不影响最终结果的场景。比如累加，最大值等，也可以用于过滤数据，在 map端将无效的数据过滤掉。

在这些需求场景下，输出的数据是可以根据key值来作合并的，合并的目的是减少输出的数据量，减少IO的读写，减少网络传输,以提高MR的作业效率。

1.combiner的作用就是在map端对输出先做一次合并,以减少传输到reducer的数据量.

2.combiner最基本是实现本地key的归并,具有类似本地reduce,那么所有的结果都是reduce完成,效率会相对降低。

3.使用combiner,先完成的map会在本地聚合,提升速度.

    --partition意思为分开，分区。它分割map每个节点的结果，按照key分别映射给不同的reduce，也是可以自定义的。其实可以理解归类。也可以理解为根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理 

partition的作用就是把这些数据归类。每个map任务会针对输出进行分区，及对每一个reduce任务建立一个分区。划分分区由用户定义的partition函数控制，默认使用哈希函数来划分分区。 HashPartitioner是mapreduce的默认partitioner。计算方法是

which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks，得到当前的目的reducer。

10、HDFS数据写入实现机制

    --写入HDFS过程：

        1、根namenode通信请求上传文件，namenode检查目标文件是否已存在，父目录是否存在 
        2、namenode返回是否可以上传 
        3、client会先对文件进行切分，比如一个blok块128m，文件有300m就会被切分成3个块，一个128M、一个128M、一个44M请求第一个 block该传输到哪些datanode服务器上 
        4、namenode返回datanode的服务器 
        5、client请求一台datanode上传数据（本质上是一个RPC调用，建立pipeline），第一个datanode收到请求会继续调用第二个datanode，然后第二个调用第三个datanode，将整个pipeline建立完成，逐级返回客户端 
        6、client开始往A上传第一个block（先从磁盘读取数据放到一个本地内存缓存），以packet为单位（一个packet为64kb），当然在写入的时候datanode会进行数据校验，它并不是通过一个packet进行一次校验而是以chunk为单位进行校验（512byte），第一台datanode收到一个packet就会传给第二台，第二台传给第三台；第一台每传一个packet会放入一个应答队列等待应答 
        7、当一个block传输完成之后，client再次请求namenode上传第二个block的服务器。

    --读取文件过程：

        使用HDFS提供的客户端开发库Client，向远程的Namenode发起RPC请求；Namenode会视情况返回文件的部分或全部block列表，对于每个block，Namenode都会返回有该block拷贝的DataNode地址；客户端开发库Client会选取离客户端最接近的DataNode来读取block；如果客户端本身就是DataNode,那么将从本地直接获取数据.读取完当前block的数据后，关闭与当前的DataNode连接，并为读取下一个block寻找最佳的DataNode；当读完列表的block后，且文件读取还没有结束，客户端开发库会继续向Namenode获取下一批的block列表。读取完一个block都会进行 checksum 验证，如果读取 datanode 时出现错误，客户端会通知 Namenode，然后再从下一个拥有该 block 拷贝的 datanode 继续读。

11、hadoop节点的动态上线下线的大概操作

    --节点上线

        1.关闭新增节点的防火墙
        2.在 NameNode节点的hosts文件中加入新增数据节点的hostname
        3.在每个新增数据节点的hosts文件中加入NameNode的hostname
        4.在NameNode节点上增加新增节点的SSH免密码登录的操作
        5.在NameNode节点上的dfs.hosts中追加上新增节点的hostname,
        6.在其他节点上执行刷新操作：hdfs dfsadmin -refreshNodes
        7.在 NameNode 节点上，更改slaves文件，将要上线的数据节点hostname追加
        到slaves文件中
        8.启动DataNode节点

        9.查看NameNode的监控页面看是否有新增加的节点      

   --节点下线

        1.修改/conf/hdfs-site.xml文件
        2.确定需要下线的机器，dfs.osts.exclude文件中配置好需要下架的机器，这个是阻
        止下架的机器去连接NameNode
        3.配置完成之后进行配置的刷新操作./bin/hadoop dfsadmin -refreshNodes,这个
        操作的作用是在后台进行block块的移动
        4.当执行三的命令完成之后，需要下架的机器就可以关闭了，可以查看现在集
        群上连接的节点，正在执行 Decommission，会显示：
        Decommission Status : Decommission in progress 执行完毕后，会显示：
        Decommission Status : Decommissioned

        5.机器下线完毕，将他们从 excludes 文件中移除。