Hadoop 学习笔记和常见问题

Question

Hadoop

• HDFS：分布式文件系统
• MAPREDUCE：分布式运算程序开发框架
• HIVE：基于大数据技术（文件系统+运算框架）的 SQL 数据仓库工具
• HBASE：基于 HADOOP 的分布式海量数据库
• ZOOKEEPER：分布式协调服务基础组件
• Flume：日志数据采集框架

集群部署

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

1. 安装 JDK 并配置环境变量（/etc/profile）
2. 关闭防火墙
3. 配置 hosts 文件，方便 hadoop 通过主机名访问（/etc/hosts）
4. 设置 ssh 免密码登录
5. 解压缩 hadoop 安装包，并配置环境变量
6. 修改配置文件（$HADOOP_HOME/conf）

hadoop-env.sh  core-site.xml  hdfs-site.xml mapred-site.xml Slaves

7. 格式化 hdfs 文件系统 （hadoop namenode -format）
8. 启动 hadoop （$HADOOP_HOME/bin/start-all.sh）
9. 使用 jps 查看进程

• hadoop-env.sh （配置 JAVA_HOME）
• core-site.xml（配置 nameNode）
• hdfs-site.xml（dfs.namenode.name.dir 、dfs.datanode.data.dir）
• mapred-site.xml（mapreduce.framework.name 配置 mapreduce 资源管理器 yarn）
• yarn-site.xml（yarn.resourcemanager.hostname）

HDFS

• HDFS 中的文件在物理上是分块存储（block），块的大小可以通过配置参数( dfs.blocksize) 来规定，默认大小在 hadoop2.x 版本中是 128M，老版本中是 64M
• HDFS 文件系统会给客户端提供一个统一的抽象目录树，客户端通过路径来访问文件，形如： hdfs://namenode :port/dir-a/dir-b/dir-c/file.data
• 目录结构及文件分块位置信息（元数据) 的管理由 namenode 节点承担--namenode 是 HDFS 集群主节点，负责维护整个 hdfs 文件系统的目录树，以及每一个路径（文件）所对应的 block 块信息（block 的 id，及所在的 datanode 服务器）
• 文件的各个 block 的存储管理由 datanode 节点承担-- datanode 是 HDFS 集群从节点，每一个 block 都可以在多个 datanode 上存储多个副本（副本数量也可以通过参数设置 dfs.replication，默认是 3）

hdfs 的工作机制

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

namenode 对数据的管理采用了三种存储形式

• 内存元数据(NameSystem)
• 磁盘元数据镜像文件
• 数据操作日志文件（可通过日志运算出元数据）

当客户端对 hdfs 中的文件进行新增或者修改操作，操作记录首先被记入 edits 日志文件中，当客户端操作成功后，相应的元数据会更新到内存 meta.data 中

HDFS 读写流程

写详细步骤：

• 1.客户端向 NameNode 发出写文件请求。
• 2.检查是否已存在文件、检查权限。若通过检查，直接先将操作写入 EditLog，并返回输出流对象。
  （注：WAL，write ahead log，先写 Log，再写内存，因为 EditLog 记录的是最新的 HDFS 客户端执行所有的写操作。如果后续真实写操作失败了，由于在真实写操作之前，操作就被写入 EditLog 中了，故 EditLog 中仍会有记录，我们不用担心后续 client 读不到相应的数据块，因为在第 5 步中 DataNode 收到块后会有一返回确认信息，若没写成功，发送端没收到确认信息，会一直重试，直到成功）
• 3.client 端按 128MB 的块切分文件。
• 4.client 将 NameNode 返回的分配的可写的 DataNode 列表和 Data 数据一同发送给最近的第一个 DataNode 节点，此后 client 端和 NameNode 分配的多个 DataNode 构成 pipeline 管道，client 端向输出流对象中写数据。client 每向第一个 DataNode 写入一个 packet，这个 packet 便会直接在 pipeline 里传给第二个、第三个…DataNode。
  （注：并不是写好一个块或一整个文件后才向后分发）
• 5.每个 DataNode 写完一个块后，会返回确认信息。
  （注：并不是每写完一个 packet 后就返回确认信息，个人觉得因为 packet 中的每个 chunk 都携带校验信息，没必要每写一个就汇报一下，这样效率太慢。正确的做法是写完一个 block 块后，对校验信息进行汇总分析，就能得出是否有块写错的情况发生）
• 6.写完数据，关闭输输出流。
• 7.发送完成信号给 NameNode。
  （注：发送完成信号的时机取决于集群是强一致性还是最终一致性，强一致性则需要所有 DataNode 写完后才向 NameNode 汇报。最终一致性则其中任意一个 DataNode 写完后就能单独向 NameNode 汇报，HDFS 一般情况下都是强调强一致性）

关于 pipline 是一种管道模式，我的理解是类似在 redis 里面的 pipline ，如果每次执行一条命令 需要这样的过程 请求数据 等待处理 返回结果 这样的来回操作很耗时 采用一种管道的概念一次发送多条命令 交给服务器去优化处理 最终将结果一次返回。

读详细步骤：

• 1.client 访问 NameNode，查询元数据信息，获得这个文件的数据块位置列表，返回输入流对象。
• 2.就近挑选一台 datanode 服务器，请求建立输入流 。
• 3.DataNode 向输入流中中写数据，以 packet 为单位来校验。
• 4.关闭输入流

Mapreduce 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 的结果文件，reducetask 会将这些文件再进行合并（归并排序）
• 7、合并成大文件后，shuffle 的过程也就结束了，后面进入 reducetask 的逻辑运算过程（从文件中取出一个一个的键值对 group，调用用户自定义的 reduce() 方法）

Shuffle 中的缓冲区大小会影响到 mapreduce 程序的执行效率，原则上说，缓冲区越大，磁盘 io 的次数越少，执行速度就越快
缓冲区的大小可以通过参数调整，参数：io.sort.mb 默认 100M）

MAPREDUCE 中的 Combiner

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

Combiner 所做的事情：
每一个 map 都可能会产生大量的本地输出，Combiner 的作用就是对 map 端的输出先做一次合并，以减少在 map 和 reduce 节点之间的数据传输量；
Combiner 的意义：
在 MapReduce 中，当 map 生成的数据过大时，带宽就成了瓶颈，当在发送给 Reduce 时对数据进行一次本地合并，
减少数据传输量以提高网络 IO 性能；
Combiner 的时机：
Combiner 最基本的是实现本地 key 的聚合，有本地 Reduce 之称，实际上是现实就继承来 Reducer ，本质上就是一个 Reducer。

请写出以下执行命令

1. 杀死一个 job? hadoop job –kill job_id
2. 删除 hdfs 上的/tmp/aaa 目录? hadoop fs -rmr hdfs_path
3. 动态添加 datanode，不停 namenode？

1.修改 slaves 文件，添加需要增加的节点 host 或者 ip，并将其更新到各个节点
2.在 datanode 中启动执行启动 datanode 命令。命令：sh hadoop-daemon.sh start datanode
3.可以通过 web 界面查看节点添加情况。或使用命令：sh hadoop dfsadmin -report
4.执行 hadoop balance 命令。（此项为 balance 集群使用，如果只是添加节点，则此步骤不需要）

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

比较流行的三种调度器有：默认调度器 FIFO，计算能力调度器 CapacityScheduler，公平调度器 Fair Scheduler

1. 默认调度器 FIFO

hadoop 中默认的调度器，采用先进先出的原则

2. 计算能力调度器 CapacityScheduler

选择占用资源小，优先级高的先执行

3. 公平调度器 FairScheduler

同一队列中的作业公平共享队列中所有资源

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

1. 内存数据库 derby，较小，不常用
2. 本地 mysql，较常用
3. 远程 mysql，不常用

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

在 Hadoop 中，默认情况下是按照 key 进行排序，如果要按照 value 进行排序怎么办？

有两种方法进行二次排序，分别为： buffer and in memory sort 和 value-to-key conversion 。

buffer and in memory sort

主要思想是：在 reduce() 函数中，将某个 key 对应的所有 value 保存下来，然后进行排序。 这种方法最大的缺点是：可能会造成 out of memory 。

value-to-key conversion

主要思想是：将 key 和部分 value 拼接成一个组合 key（实现 WritableComparable 接口或者调 setSortComparatorClass 函数），这样 reduce 获取的结果便是先按 key 排序，后按 value 排序的结果，需要注意的是，用户需要自己实现 Paritioner，以便只按照 key 进行数据划分。

简述 hadoop 实现 join 的几种方法？

reduce side join

reduce side join 是一种最简单的 join 方式，其主要思想如下：

在 map 阶段，map 函数同时读取两个文件 File1 和 File2 ，为了区分两种来源的 key/value 数据对，对每条数据打一个标签（tag），比如： tag=0 表示来自文件 File1 ， tag=2 表示来自文件 File2 。即：map 阶段的主要任务是对不同文件中的数据打标签。

在 reduce 阶段，reduce 函数获取 key 相同的来自 File1 和 File2 文件的 value list， 然后对于同一个 key，对 File1 和 File2 中的数据进行 join（笛卡尔乘积）。即：reduce 阶段进行实际的连接操作。

map side join

之所以存在 reduce side join，是因为在 map 阶段不能获取所有需要的 join 字段，即：同一个 key 对应的字段可能位于不同 map 中。Reduce side join 是非常低效的，因为 shuffle 阶段要进行大量的数据传输。

Map side join 是针对以下场景进行的优化：两个待连接表中，有一个表非常大，而另一个表非常小，以至于小表可以直接存放到内存中。这样，我们可以将小表复制多份，让每个 map task 内存中存在一份（比如存放到 hash table 中），然后只扫描大表：对于大表中的每一条记录 key/value，在 hash table 中查找是否有相同的 key 的记录，如果有，则连接后输出即可。

为了支持文件的复制，Hadoop 提供了一个类 DistributedCache，使用该类的方法如下：

（1）用户使用静态方法 DistributedCache.addCacheFile() 指定要复制的文件，它的参数是文件的 URI（如果是 HDFS 上的文件，可以这样： hdfs://namenode:9000/home/XXX/file ，其中 9000 是自己配置的 NameNode 端口号）。JobTracker 在作业启动之前会获取这个 URI 列表，并将相应的文件拷贝到各个 TaskTracker 的本地磁盘上。（2）用户使用 DistributedCache.getLocalCacheFiles() 方法获取文件目录，并使用标准的文件读写 API 读取相应的文件。

SemiJoin

SemiJoin，也叫半连接，是从分布式数据库中借鉴过来的方法。它的产生动机是：对于 reduce side join，跨机器的数据传输量非常大，这成了 join 操作的一个瓶颈，如果能够在 map 端过滤掉不会参加 join 操作的数据，则可以大大节省网络 IO。

实现方法很简单：选取一个小表，假设是 File1，将其参与 join 的 key 抽取出来，保存到文件 File3 中，File3 文件一般很小，可以放到内存中。在 map 阶段，使用 DistributedCache 将 File3 复制到各个 TaskTracker 上，然后将 File2 中不在 File3 中的 key 对应的记录过滤掉，剩下的 reduce 阶段的工作与 reduce side join 相同。

更多关于半连接的介绍，可参考：半连接介绍。

reduce side join + BloomFilter

在某些情况下，SemiJoin 抽取出来的小表的 key 集合在内存中仍然存放不下，这时候可以使用 BloomFiler 以节省空间。

BloomFilter 最常见的作用是：判断某个元素是否在一个集合里面。它最重要的两个方法是：add() 和 contains()。最大的特点是不会存在 false negative，即：如果 contains() 返回 false，则该元素一定不在集合中，但会存在一定的 true negative，即：如果 contains() 返回 true，则该元素可能在集合中。

因而可将小表中的 key 保存到 BloomFilter 中，在 map 阶段过滤大表，可能有一些不在小表中的记录没有过滤掉（但是在小表中的记录一定不会过滤掉），这没关系，只不过增加了少量的网络 IO 而已。

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

1. 内存数据库 derby，较小，不常用
2. 本地 mysql，较常用
3. 远程 mysql，不常用