这份笔记整合了 Hadoop 核心组件的底层原理，聚焦 Hadoop 3.x 知识体系。重点讲清楚 HDFS、MapReduce、YARN 的工作机制，以及 3.x 版本带来的关键变化。

---

一、Hadoop 核心概述

1.1 Hadoop 到底是什么

Hadoop 是一套开源的分布式基础架构，核心解决两个问题：

• 海量数据存储 → HDFS（Hadoop Distributed File System）
• 海量数据计算 → MapReduce

资源调度由 YARN 负责，Common 提供底层工具支持。Hadoop 的设计哲学很简单：用廉价的普通机器组成集群，通过横向扩展来扛住 PB 级别的数据。

1.2 四大核心组件

组件	职责	类比理解
Hadoop Common	底层工具库，为其他模块提供支持	基础设施
HDFS	分布式文件系统，负责数据存储	分布式硬盘
MapReduce	分布式计算框架，负责数据处理	分布式 CPU
YARN	资源调度平台，负责集群资源分配	分布式操作系统

1.3 Hadoop 3.x 核心特性概述

当前生产环境主流是 3.x 系列（最新稳定版 3.4.2，2025 年 8 月发布）。这套版本几个关键特性需要掌握：

• 最低 JDK 要求：3.x 最低 JDK 8，低于这个版本的编译和运行都不支持
• HDFS 纠删码：默认支持 Erasure Coding，存储开销从传统 3 副本的 300% 降到 150%
• 多 NameNode HA：不再局限于 1 个 Active + 1 个 Standby，可以配多个 Standby，容错等级更高
• YARN 新能力：支持 GPU 调度、节点标签、Opportunistic Containers、Docker/K8s 集成
• 默认块大小：Hadoop 3.x 默认 128MB，这个参数从上一代延续过来，没有变化

---

二、HDFS 分布式文件系统

2.1 为什么需要分布式文件系统

单台机器的存储和吞吐量有天花板，分布式文件系统的核心思路是：把大文件切成块，分散存到多台机器上，对外暴露一个统一的文件系统视图。

客户端不需要关心数据具体存在哪台机器，就像操作本地磁盘一样操作分布式集群。

2.2 HDFS 核心架构

HDFS 是典型的 Master-Slave 架构：

• NameNode（主节点）：管理整个文件系统的命名空间和元数据，不存实际数据
• DataNode（从节点）：存储实际的文件数据块（Block），定期向 NameNode 汇报
• SecondaryNameNode：历史遗留的辅助组件，负责合并 FsImage 和 EditLog（仅在非 HA 模式下存在）

Hadoop 3.x 生产环境标配 HA 架构，核心变化是把 SecondaryNameNode 换成了 JournalNode 集群 + ZKFC（ZooKeeper Failover Controller）：

• JournalNode（通常 3 个）同步 Active 和 Standby NameNode 的元数据日志
• ZKFC 监控 NameNode 状态，负责主备自动切换
• Hadoop 3.x 支持多个 Standby NameNode，进一步提升容错能力

2.3 Block 块机制

HDFS 的存储单元是 Block，所有文件都会被切分成固定大小的 Block 来存储。

默认块大小：128MB（dfs.blocksize 配置，单位字节，默认 134217728）

副本机制：默认每个 Block 存 3 份（dfs.replication），这也是 HDFS 保证容错的核心手段。

关于 128MB 怎么来的：NameNode 的元数据全部存在内存里。如果块太小（比如 4KB），1TB 数据会产生 2.5 亿个块，NameNode 直接崩掉。如果块太大（比如 1GB），MapReduce 的并行度又会受限制。128MB 是在元数据压力和计算并行度之间取的一个平衡点。

机架感知（Rack Awareness）

3 副本不是随机存的，HDFS 有一套机架感知策略来降低数据丢失风险：

• 第 1 个副本：放在客户端所在的本地节点（如果客户端在集群外，就随机选一个节点）
• 第 2 个副本：放到与第 1 个副本不同机架的节点上
• 第 3 个副本：放到与第 2 个副本相同机架的其他节点上
• 更多副本：随机选择

这样设计的逻辑是：同一个机架内的节点挂了（比如交换机故障）不会同时影响两份副本。

踩坑点：块大小设置

块大小不是越大越好。如果你的数据集主要是几十 MB 的小文件，强行切成 128MB 的块反而浪费空间（一个块只存几 MB，NameNode 元数据却按一个完整块来记）。反过来，如果你跑的是 TeraSort 这种大数据量竞赛任务，把块调到 256MB 甚至 512MB 能减少 MapTask 数量，降低调度开销。

生产环境还有一个隐藏坑：如果 HDFS 上存了大量小文件（< 块大小的一半），NameNode 的内存会被元数据撑爆。1000 万个文件大概需要 NameNode 8~10GB 内存，这是很多人扩容集群时忽略的点。发现 NameNode 内存告警，先 hdfs dfs -count 看看文件数量是不是失控了。

2.4 HDFS 写数据流程

客户端往 HDFS 写文件，底层经历以下步骤：

1. 客户端调用 DistributedFileSystem.create()，向 NameNode 发起创建文件请求
2. NameNode 校验权限和目标文件是否存在，校验通过后返回可以写入的 DataNode 列表
3. 客户端与第一个 DataNode 建立 Pipeline（TCP 管道连接），DataNode-1 再连 DataNode-2，DataNode-2 再连 DataNode-3
4. 数据以 Packet（默认 64KB）为单位，先写入客户端的 DataStreamer 本地缓冲区，然后推入 Pipeline
5. 每个 DataNode 收到 Packet 后，先写本地磁盘，再转发给 Pipeline 的下一个节点
6. 写完一个 Block 后，从最后一个 DataNode 开始反向发送 ACK 确认，最终回到客户端
7. 客户端通知 NameNode，这个 Block 写完了，NameNode 更新元数据

关键细节

• DataStreamer 缓冲队列：客户端不是等 DataNode 写完才发下一个 Packet，而是维护一个 packets 队列来异步发送，Pipeline 里可以同时有多个 Packet 在传输
• ACK 队列：Packet 发出去后会进入 ACK 队列，收到 ACK 后才从队列移除
• 本地校验：每个 DataNode 写完 Packet 后会做 CRC 校验，确保数据没损坏
• 失败处理：Pipeline 中某个 DataNode 挂了，客户端会立刻把失败的 Packet 加到队列前端，重新选一个 DataNode 重建 Pipeline
• 副本系数：写的时候如果传了 replication 参数，以传的为准；没传就用 dfs.replication 默认值

2.5 HDFS 读数据流程

读数据比写简单，核心逻辑是就近拉取：

1. 客户端调用 open()，向 NameNode 请求读取文件
2. NameNode 返回文件的元数据，包括这个文件由哪些 Block 组成、每个 Block 在哪些 DataNode 上
3. 客户端优先选择本地节点读取，如果本地没有，就选同机架的节点，最后才选跨机架
4. 串行读取各个 Block，读取过程中会做 checksum 校验
5. 一个 Block 读完才读下一个，读完所有 Block 后关闭连接

2.6 DataNode 工作机制

心跳与块汇报

DataNode 启动后会做两件事：

• 心跳：每 3 秒（dfs.heartbeat.interval）向 NameNode 发一次心跳，告诉 NameNode “我还活着”，同时上报自己的存储容量、剩余空间、正在处理的块等信息
• 块汇报：DataNode 启动时会做一次全量块汇报，之后定期做增量汇报，让 NameNode 知道它手上都有哪些 Block 副本

心跳超时判定：NameNode 如果连续 10 分钟（默认 timeout = 10 * heartbeat.recheck.interval + 2 * heartbeat.interval，即 10 分 30 秒）收不到某个 DataNode 的心跳，就认定这个节点挂了。

节点挂了之后，NameNode 会检查这个节点上的 Block 副本是否低于设定的副本数，如果不够就安排其他 DataNode 重新复制，保证副本数恢复到目标值。

踩坑点：DataNode 假死

DataNode 进程没挂但处理心跳的线程卡住了（比如磁盘 IO 饱和），这种情况 NameNode 照样会在 10 分钟后把它标记为 dead，然后疯狂触发复制。等 DataNode 恢复正常，NameNode 又会发现副本多了，再触发删除。这一来一回会产生大量网络 IO，拖垮集群。

排查命令：hdfs dfsadmin -report 看 Live/Dead 节点数是不是在正常波动；jstack 抓 DataNode 线程栈，看看 heartbeat 线程是不是被锁住了。如果是磁盘问题，先把 dfs.datanode.data.dir 挂的那块盘从挂载点剔除，再重启 DataNode。

2.7 安全模式（SafeMode）

NameNode 启动时会自动进入安全模式，此时文件系统处于只读状态，不接受写操作（创建、删除、修改都不行）。

进入安全模式的时机：

• NameNode 启动时：需要加载 FsImage + 重放 EditLog，然后等待 DataNode 上报块信息，确认副本满足最小要求
• 手动进入：hdfs dfsadmin -safemode enter
• 触发阈值：当可用 Block 的比例低于 dfs.namenode.safemode.threshold-pct（默认 0.999f，即 99.9%）时，NameNode 会自动进入安全模式

退出条件：DataNode 报告的最小副本数满足阈值的 Block 达到设定比例，并且持续时间超过 dfs.namenode.safemode.extension（默认 30 秒）。

hdfs dfsadmin -safemode get      # 查看当前状态
hdfs dfsadmin -safemode leave    # 强制退出（慎用，可能数据还没汇报完）

踩坑点：安全模式卡住

集群重启后如果大量 DataNode 没起来，或者 JournalNode 数据不同步，NameNode 可能一直卡在安全模式。这时候先别急着强制 leave，先看看 hdfs dfsadmin -report 里 Missing/Replicated 的块是不是异常。如果是 JournalNode 挂了导致元数据不同步，先修 JournalNode，让 Standby 和 Active 的 EditLog 对齐。

节点退役与重新上线

生产环境下线节点不能直接把机器关机，否则 NameNode 会在 10 分钟后才感知到节点挂了，期间这个节点上的 Block 会被认为不可达，触发大量副本复制，等复制完了再把节点标记为 dead，完全是浪费资源。

正确做法是把节点加入 dfs.hosts.exclude 黑名单，然后让 NameNode 平滑退役：

# 1. 在 NameNode 的 hdfs-site.xml 里配置 exclude 文件路径
<property>
  <name>dfs.hosts.exclude</name>
  <value>/etc/hadoop/conf/dfs.exclude</value>
</property>

# 2. 把要下线的节点 hostname 写进 /etc/hadoop/conf/dfs.exclude
echo "node-to-decommission" >> /etc/hadoop/conf/dfs.exclude

# 3. 让 NameNode 重新加载配置（不用重启）
hdfs dfsadmin -refreshNodes

然后盯一下退役进度：

hdfs dfsadmin -report | grep "Decommission"

等状态变成 Decommissioned 就可以安全关机了。重新上线时从 exclude 文件里删掉这行，refreshNodes 即可。

2.8 HDFS 联邦（Federation）

HDFS 联邦出现得比较早，在 Hadoop 2.x 时代就已经引入，解决的是单个 NameNode 内存瓶颈问题。

核心思路：一个集群里可以有多个独立的 NameNode，每个 NameNode 管理独立的命名空间（Namespace），但它们共享底层同一批 DataNode 的存储池。

举例：你有 2000 台 DataNode 的集群，NameNode-1 管理 /warehouse 目录，NameNode-2 管理 /logs 目录，NameNode-3 管理 /tmp 目录。三个 NameNode 各自维护自己的 FsImage 和 EditLog，DataNode 上的 Block 会被所有 NameNode 共享调度。

客户端通过 ViewFS 来访问联邦集群，配置文件里用 mount table 把不同路径映射到不同的 NameNode。

联邦 vs HA 的区别

特性	HA	Federation
解决的问题	NameNode 单点故障	NameNode 内存/性能瓶颈
NameNode 数量	2 ~ N 个（同一命名空间）	多个（各管各的命名空间）
DataNode	独立集群	共享同一批 DataNode
元数据	完全同步	完全独立

生产环境通常是 HA + Federation 混用：每个联邦命名空间内部再配 HA。

2.9 短路读（Short Circuit Local Reads）

正常情况下，即使 Client 和 DataNode 在同一台机器上，读取数据也要走 TCP socket：Client → DataNode → 本地磁盘。

短路读让客户端绕过 DataNode 进程，直接读取本地磁盘上的 Block 文件，节省一次 socket 通信和一次进程上下文切换。

启用条件：

• Client 和 DataNode 在同一台物理机上
• 配置 dfs.client.read.shortcircuit = true
• 需要配置共享内存 dfs.domain.socket.path（Unix Domain Socket）
• 客户端用户必须在 dfs.block.local-path-access.user 允许的列表里

性能提升：对随机读密集型任务（比如 HBase）提升非常明显，能减少 30% 以上的读取延迟。

2.10 HDFS 负载均衡（Balancer）

DataNode 之间存储不均匀是常见问题：

• 新节点上线时空盘，老节点快满了
• 某些节点硬盘容量比其他节点大
• 写入数据时副本分配不均

HDFS 提供了 Balancer 工具来做数据重分布：

# 启动 balancer，默认阈值 10%（即每个 DataNode 的使用率与集群平均使用率的差值不超过 10%）
start-balancer.sh -threshold 10

# 指定带宽限制（默认 1MB/s，防止打满网络）
hdfs dfsadmin -setBalancerBandwidth 104857600

Hadoop 3.x 还引入了 Disk Balancer，解决的是单节点内部多块磁盘之间的不均衡。比如一台机器挂了 4 块 8TB 盘，其中一块盘比其他盘满了，用 Disk Balancer 来做 intra-node 级别的迁移。

hdfs diskbalancer -plan node1   # 生成均衡计划
hdfs diskbalancer -execute node1.plan.json

踩坑点：Balancer 跑不动

Balancer 默认只迁移那些与平均值偏差超过阈值的数据块。如果集群整体写入速度比 balancer 迁移速度快，它永远追不平。另外，Balancer 不会迁移正在写入的块，所以高峰期启动 balancer 效果很差。建议在业务低峰期跑，并且把带宽限制调高一些。

2.11 Trash 回收站与快照

Trash 回收站：HDFS 默认开启了回收站机制（fs.trash.interval，默认 0 即关闭，生产环境建议配成 1440 分钟即 1 天）。删除的文件不会立即消失，而是先移到 /user/<username>/.Trash 里，过期后才真正删除。

# 立即彻底删除（跳过回收站）
hdfs dfs -rm -skipTrash /path/to/file

# 恢复误删文件（从 Trash 移回原位置）
hdfs dfs -mv /user/admin/.Trash/Current/path/to/file /path/to/file

快照（Snapshot）：HDFS 支持目录级别的快照，冻结某一时刻的数据状态。快照是只读的，且创建时几乎是 O(1) 操作（写时复制）。

hdfs dfsadmin -allowSnapshot /path/to/dir    # 先开启目录的快照功能
hdfs dfs -createSnapshot /path/to/dir snap1  # 创建快照
hdfs dfs -ls /path/to/dir/.snapshot/snap1    # 查看快照内容

快照的典型用途：数据仓库 ODS 层每日全量同步前，先对源目录打快照，如果同步失败可以快速回滚。

2.12 HDFS 配额（Quota）

HDFS 支持对目录设置名称配额和空间配额：

• 名称配额（Name Quota）：限制目录下文件/子目录的数量
• 空间配额（Space Quota）：限制目录下文件总大小（含副本）

hdfs dfsadmin -setQuota 1000 /user/data    # 最多 1000 个子项
hdfs dfsadmin -setSpaceQuota 1t /user/data  # 最多 1TB（含副本）

超出配额后写入会报 NSQuotaExceededException 或 DSQuotaExceededException。这在多租户共享集群里很有用，防止某个部门把集群写爆。

2.13 元数据管理与 HA

元数据的组成

NameNode 内存里维护两类核心数据：

• FsImage：文件系统命名空间的完整快照（文件/目录的树状结构、权限、修改时间等持久化信息）
• EditLog：每次写操作（创建、删除、重命名）的增量日志

NameNode 启动时把 FsImage 加载到内存，再重放 EditLog，就恢复了最新的文件系统状态。

Checkpoint 机制

FsImage + EditLog 会越来越庞大，所以需要定期合并：

• SecondaryNameNode：每隔一段时间（默认 1 小时或 EditLog 达到 64MB）把 NameNode 的 FsImage 和 EditLog 拉过来合并，生成新的 FsImage，再推回给 NameNode
• HA 模式（新版）：由 Standby NameNode 负责做 Checkpoint，不需要 SecondaryNameNode 了

HA 自动切换

生产环境必须配 HA，避免 NameNode 单点故障导致整个集群不可用：

• JournalNode 集群（至少 3 个，奇数个）：Active NameNode 的每次写操作都同步写到 JournalNode，Standby NameNode 实时从 JournalNode 读 EditLog 保持元数据同步
• ZKFC：每个 NameNode 节点上跑一个 ZKFC 进程，通过 ZooKeeper 做分布式锁选举。Active 挂了，ZKFC 自动把 Standby 提升为 Active
• 脑裂防护：通过 fencing 机制（比如 SSH 杀掉旧 Active 的进程）防止出现两个 NameNode 同时认为自己是 Active 的情况

2.14 Hadoop 3.x 纠删码（Erasure Coding）

这是 3.x 最重要的存储层升级。传统 3 副本的存储开销是 300%，纠删码可以降到 150%。

原理不复杂：把数据切分成条带（strip），根据 Reed-Solomon 算法生成校验块。数据块和校验块一起分散存储，即使部分块丢失也能通过算法恢复原始数据。

HDFS 内置的策略：

策略	数据块	校验块	存储开销	容错能力
RS-6-3（默认）	6	3	150%	允许丢 3 个块
RS-3-2	3	2	166%	允许丢 2 个块
RS-10-4	10	4	140%	允许丢 4 个块
XOR-2-1	2	1	150%	允许丢 1 个块

使用方式：纠删码策略是目录级别的配置，新建文件继承父目录的策略。不能对单个已有文件直接改策略，需要 distcp 复制到新目录。

# 查看支持的策略
hdfs ec -listPolicies

# 对某个目录启用 RS-6-3
hdfs ec -enablePolicy -policy RS-6-3-1024k
hdfs ec -setPolicy -path /warehouse/data -policy RS-6-3-1024k

纠删码对 CPU 和网络有额外消耗（编解码计算 + 跨机架读取），所以适合大文件、冷数据、一次写入多次读取的场景。热数据、小文件、频繁随机写的场景还是用传统副本更合适。

2.15 HDFS 常用 Shell 命令

# 上传文件
hdfs dfs -put local.txt /hdfs/path
hdfs dfs -copyFromLocal local.txt /hdfs/path

# 下载文件
hdfs dfs -get /hdfs/path ./local.txt

# 查看目录
hdfs dfs -ls /
hdfs dfs -lsr /   # 递归（新版推荐用 -ls -R）

# 创建/删除目录
hdfs dfs -mkdir -p /test/dir
hdfs dfs -rm -r /test    # -r 递归删除，-skipTrash 不进回收站

# 查看文件内容
hdfs dfs -cat /test/a.txt
hdfs dfs -tail /test/a.log

# 移动/重命名
hdfs dfs -mv /src /dst

# 文件信息
hdfs dfs -stat "file size=%b,blocks=%o" /test/a.txt

# 手动触发块汇报（运维排障常用）
hdfs dfsadmin -triggerBlockReport datanode_host:port

---

三、MapReduce 分布式计算框架

3.1 MapReduce 的设计思想

MapReduce 是 Google 提出的编程模型，Hadoop 做了开源实现。核心思路就一句话：分而治之。

• Map 阶段：把大问题拆成小问题，分发到各个节点并行处理
• Reduce 阶段：汇总各个节点的处理结果，得到最终答案

它最适合的场景是批量离线处理，比如日志分析、数据清洗、聚合统计。实时计算别用 MapReduce，延迟太高。

3.2 完整执行流程

一个 MapReduce Job 的执行包含三个阶段：Map → Shuffle → Reduce。

Map 阶段

1. InputFormat & RecordReader：按 InputSplit（默认 128MB，一个 Block 对应一个 Split）读取输入数据，每读一条记录就调用一次 map()
2. Mapper：用户自定义的 map() 逻辑，输出 <key, value> 键值对
3. 环形缓冲区：Mapper 的输出先写入内存中的环形缓冲区（默认 100MB），而不是直接写磁盘

Shuffle 阶段（核心中的核心）

Shuffle 是 Map 输出到 Reduce 输入之间的桥梁，也是面试最爱问的部分。

Map 端 Shuffle：

1. 数据进入环形缓冲区后，先分区（默认 HashPartitioner，key.hashCode() % numReduceTasks）
2. 每个分区内部按 key 做快速排序
3. 当缓冲区数据量达到阈值（默认 80%）时，触发溢写（Spill），把排好序的数据刷到磁盘，生成临时文件
4. 所有 Map 任务完成后，对磁盘上的多个溢写文件做归并排序（Merge），合并成一个最终文件

设置 mapreduce.map.sort.spill.percent=0.80 控制溢写阈值。设置 mapreduce.task.io.sort.mb=100 控制缓冲区大小（单位 MB）。

Reduce 端 Shuffle：

1. ReduceTask 启动后，主动向各个 MapTask 拉取属于自己分区的数据（fetch）
2. 拉取过来的数据先存内存，不够了溢写到磁盘
3. 对所有拉取的数据做归并排序，确保同一个 key 的数据聚在一起
4. 按 key 分组，交给 reduce() 处理

Reduce 阶段

1. Reducer：用户自定义的 reduce() 逻辑，对同一 key 的所有 value 做汇总计算
2. OutputFormat：把结果输出到 HDFS（默认 TextOutputFormat，每行一个 <key\tvalue>）

3.3 并行度怎么定

MapTask 数量

MapTask 个数 = InputSplit 个数 = 输入文件总大小 / BlockSize（默认 128MB）

举例：输入目录有 150MB 和 100MB 两个文件，那么 Split 数 = ceil(150/128) + ceil(100/128) = 2 + 1 = 3 个 MapTask。

FileInputFormat.setMinInputSplitSize() 和 setMaxInputSplitSize() 可以调整，但一般不动。

ReduceTask 数量

由用户代码显式设置：

job.setNumReduceTasks(4);

• 0 个 ReduceTask：没有 Reduce 阶段，Map 输出直接写 HDFS（适合过滤、转换类操作）
• 1 个 ReduceTask：所有数据汇总到一个节点（容易 OOM）
• 建议根据 key 的分布来设置，一般几十个到几百个不等，太少会数据倾斜，太多会调度开销过大

3.4 自定义组件

自定义 Partitioner

默认的 HashPartitioner 有时候会导致数据倾斜。你可以按业务需求自定义：

public class CustomPartitioner extends Partitioner<Text, LongWritable> {
    @Override
    public int getPartition(Text key, LongWritable value, int numPartitions) {
        // 比如按手机号前三位分区
        String prefix = key.toString().substring(0, 3);
        return (prefix.hashCode() & Integer.MAX_VALUE) % numPartitions;
    }
}

job.setPartitionerClass(CustomPartitioner.class);
job.setNumReduceTasks(5);

自定义 Partitioner 后，要确保 numReduceTasks 和分区逻辑匹配，否则数据可能进错 ReduceTask。

Combiner（局部聚合）

Combiner 是 Map 端的局部 Reduce，作用是减少 Map 到 Reduce 之间的数据传输量。

比如 WordCount，每个 MapTask 本地先做一次词频汇总，只把汇总结果发给 ReduceTask，而不是把每条原始记录都发过去。

job.setCombinerClass(WordCountReducer.class);

Combiner 不是必选项，而且只能用于满足结合律的操作（比如求和、计数）。求平均值就不能用 Combiner，因为局部平均值的平均值不等于全局平均值。

GroupingComparator（分组比较器）

ReduceTask 默认把 key 相同的记录分到一组。如果你想让不同的 key 当作同一组来处理（比如按省份汇总，同一省份的不同城市算一组），就要自定义 GroupingComparator：

public class ProvinceGroupingComparator extends WritableComparator {
    protected ProvinceGroupingComparator() {
        super(OrderBean.class, true);
    }
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        OrderBean o1 = (OrderBean) a;
        OrderBean o2 = (OrderBean) b;
        return o1.getProvinceId().compareTo(o2.getProvinceId());
    }
}

job.setGroupingComparatorClass(ProvinceGroupingComparator.class);

GroupingComparator 决定的是哪些记录进同一个 reduce 调用，而 SortComparator 决定的是排序顺序。两者可以独立设置。

3.5 Hadoop 序列化：Writable 接口

Java 自带的 Serializable 太重了，Hadoop 自己搞了一套轻量级序列化机制 Writable。

核心接口只有两个方法：

void write(DataOutput out) throws IOException;     // 序列化
void readFields(DataInput in) throws IOException;  // 反序列化

常用类型对照表：

Java 类型	Hadoop Writable 类型
boolean	BooleanWritable
byte	ByteWritable
int	IntWritable
long	LongWritable
float	FloatWritable
double	DoubleWritable
String	Text
byte[]	BytesWritable
null	NullWritable

自定义对象实现 Writable 示例：

public class FlowBean implements Writable {
    private long upFlow;
    private long downFlow;
    private long sumFlow;

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();
    }

    @Override
    public String toString() {
        return upFlow + "\t" + downFlow + "\t" + sumFlow;
    }
}

序列化和反序列化的字段顺序必须完全一致，不然数据会错位。

3.6 Job 提交流程

本地模式

小数据量测试时常用本地模式，不提交到 YARN：

1. waitForCompletion() 触发提交
2. 通过 CreateTmpDir 创建临时工作目录（默认 /tmp/hadoop-<username>）
3. 获取本地 jobID
4. 在临时目录下创建 job.xml（所有配置参数）和 job.jar（程序包）
5. 启动 LocalJobRunner，在本地 JVM 里顺序/并行执行 MapTask 和 ReduceTask
6. 执行状态通过 JobStatus 实时更新

YARN 集群模式

生产环境的正式提交流程：

1. waitForCompletion() → submit() → connect()，建立与 YARN 的连接
2. 向集群申请 JobID（ApplicationID，格式 job_<timestamp>_<id>）
3. 把 Job 运行所需的资源（jar 包、配置文件、依赖）拷贝到 HDFS 的临时目录（/tmp/hadoop-yarn/staging）
4. 计算 InputSplit 数量，生成切片规划文件
5. 真正向 ResourceManager 提交 Job：rmClient.submitApplication()
6. ResourceManager 把 Application 加入调度队列
7. 某个空闲 NodeManager 领取到该 Job，在其 Container 里启动 MRAppMaster（MapReduce 的 ApplicationMaster）
8. MRAppMaster 下载 HDFS 上的 jar 和配置，计算需要多少个 MapTask 和 ReduceTask
9. MRAppMaster 向 ResourceManager 申请运行这些 Task 所需的 Container
10. ResourceManager 将任务分配给各个 NodeManager
11. NodeManager 收到任务后，在本地启动 Container，执行 MapTask 或 ReduceTask
12. MRAppMaster 实时监控所有 Task 的状态，失败的任务会重新申请资源重跑
13. 所有 Task 完成后，MRAppMaster 向 ResourceManager 注销自己，释放资源

第 10 步体现了 “数据向计算靠拢” 的本地化调度原则：ResourceManager 会尽量把任务分配给存储有对应 InputSplit 的 DataNode，减少网络传输。

3.7 Hadoop 3.x MapReduce 优化

Hadoop 3.x 在 MapReduce 侧做了原生性能优化，最值得关注的是 MapTask 级别的 Native Output Collector（基于 JNI）。

对于 Shuffle 密集型的任务（大量中间数据需要排序、溢写），这个优化能提升 30% 以上 的性能。核心改进是把 map 输出的排序、溢写、IFile 序列化等操作放到 native 代码里执行，减少了 Java 层的开销。

3.8 数据本地化（Data Locality）

MapReduce 调度任务时，YARN 会尽量让任务在存有输入数据的节点上执行，而不是把数据拉到任务所在的节点。这就是数据本地化，能大幅减少网络 IO。

本地化有三个级别：

• Data Local：MapTask 运行在存有输入 Split 的 DataNode 上（最优）
• Rack Local：MapTask 运行在输入 Split 所在机架的其他节点上（次优，只需要跨节点，不跨机架）
• Off Rack：MapTask 运行在其他机架的节点上（最差，需要跨机架传输数据）

数据本地化率可以在 YARN UI 上查看。如果大量任务处于 Off Rack 状态，说明集群负载过高或者数据分布严重不均，需要跑 Balancer 或者加节点。

踩坑点："数据本地化"假象

有些新手看到 MapTask 都显示 DATA_LOCAL 就以为没问题，但实际上如果该节点的 CPU/内存已经被其他任务占满了，YARN 会把这个任务推迟或者降级到别的节点。更隐蔽的坑是：HDFS 的 3 副本意味着一个 Split 理论上可以本地化到 3 个节点，但如果这 3 个节点都满了，就只能走 Rack Local 甚至 Off Rack。

3.9 推测执行（Speculative Execution）

集群里上百个 MapTask 一起跑，总会有那么几个"拖后腿"的（可能因为那台机器磁盘坏了、网络抖动、或者其他进程抢占了资源）。推测执行的机制是：如果某个 Task 的进度明显落后于同批次其他 Task（落后超过 mapreduce.job.speculative.slowtaskthreshold 默认 1.0，即进度差值超过平均值的 1 倍），YARN 会在另一个节点上启动一个备份 Task，谁先跑完就用谁的结果，另一个被杀掉。

<!-- mapred-site.xml -->
<property>
  <name>mapreduce.map.speculative</name>
  <value>true</value>
</property>
<property>
  <name>mapreduce.reduce.speculative</name>
  <value>false</value>  <!-- Reduce 阶段通常不建议开，两个 Reduce 同时写 HDFS 可能造成冲突 -->
</property>

踩坑点：推测执行导致数据重复

如果你的 Reduce 逻辑不是幂等的（比如往数据库里做 INSERT 而不是 INSERT IGNORE），推测执行就会产生重复数据。Reduce 阶段开推测执行要格外谨慎，或者确保输出是幂等的（HDFS 文件写是原子性的，同一路径的重复写不会导致脏数据，但如果是往外部系统写就要小心了）。

3.10 分布式缓存（DistributedCache）

MapReduce 任务有时候需要用到一些小文件（比如字典表、配置文件、IP 地理库），这些小文件不适合当作输入数据分片，而是需要让每个 Task 都能访问到。

DistributedCache 就是解决这个问题的：你在 Job 提交时把文件加到缓存里，YARN 会自动把这些文件分发到每个运行 Task 的节点上。

// 把 HDFS 上的文件加到分布式缓存
job.addCacheFile(new URI("hdfs://namenode:9000/dict/area_dict.txt#area_dict"));

// Mapper 里读取
File file = new File("area_dict");
BufferedReader reader = new BufferedReader(new FileReader(file));

注意 #area_dict 是给文件起的别名（symlink），这样 Task 就能通过相对路径直接访问，不需要写死 HDFS 路径。

踩坑点：缓存文件太大

DistributedCache 只适合小文件（几十 MB 以内）。如果你塞了个几百 MB 的文件进去，每个节点都复制一份，不但启动慢，还占磁盘。大文件应该用 HDFS 路径直接读，或者用 join 的方式处理。

3.11 MapReduce Join

实际业务中经常需要把两份数据集按某个 key 关联起来。MapReduce 里实现 Join 有三种常见方式：

Reduce Join（最通用）

两份数据都作为输入，Map 阶段给每条记录打上"来自哪个表"的标记，按关联 key 分区。Reduce 阶段同一个 key 的所有记录会到同一个 ReduceTask，在 reduce() 里做关联。

缺点：Shuffle 阶段要把两份数据都搬来搬去，网络开销大。

Map Join（性能最好）

如果其中一张表非常小（能放进内存），可以用 DistributedCache 把它分发到每个 MapTask，在 Map 阶段直接做关联，不需要 Reduce 阶段。

job.setNumReduceTasks(0);  // 不需要 Reduce
job.addCacheFile(new URI("hdfs:///small_table.txt#small"));

Semi Join（折中方案）

大表 join 大表时，先扫描小表提取所有关联 key，Broadcast 到 Map 阶段做过滤，只有命中 key 的大表记录才会进入 Reduce，减少 Shuffle 数据量。

3.12 计数器（Counters）

MapReduce 提供了全局计数器机制，用来统计跨 Task 的汇总指标。

// Mapper 或 Reducer 里
Counter counter = context.getCounter("myGroup", "error_record");
counter.increment(1);

Job 运行完后，你可以在控制台或 YARN UI 上看到每个计数器的总和。

计数器的典型用途：统计脏数据条数、空值个数、异常类型分布等。计数器值会上报到 AM，再由 AM 汇总给 RM，所以不适合高频调用（每来一条记录就 increment 一次会很慢）。建议批量增量，比如每处理 1000 条记录统一 increment 一次。

3.13 数据压缩

MapReduce 中间数据和最终输出都可以压缩，省磁盘、省网络。

压缩格式	是否可切片	压缩率	速度	特点
Snappy	否	中	极快	Google 出品，Hadoop 默认推荐
LZ4	否	中	极快	解压速度比 Snappy 还快
Gzip	否	高	慢	不可切片，适合最终输出
Bzip2	是	高	很慢	唯一可切片的压缩格式，但 CPU 开销大
Zstandard	是（需编解码器）	高	快	Facebook 出品，综合性能最好

配置方式：

// Map 输出压缩（减少 Shuffle 数据量）
job.getConfiguration().setBoolean("mapreduce.map.output.compress", true);
job.getConfiguration().setClass("mapreduce.map.output.compress.codec", SnappyCodec.class, CompressionCodec.class);

// 最终输出压缩
FileOutputFormat.setCompressOutput(job, true);
FileOutputFormat.setOutputCompressorClass(job, SnappyCodec.class);

踩坑点：压缩不可切片

如果你用 Gzip/Snappy 压缩了一个 2GB 的文本文件，MapReduce 只能把它当成一个 InputSplit（因为不支持随机定位压缩边界），结果就是 1 个 MapTask 吭哧吭哧跑半天，其他核都在闲着。大文件要用 Bzip2 或者先切分再压缩。

3.14 Uber 模式

Uber 模式是 YARN 针对小 Job 做的优化。如果 Job 的数据量很小（比如只有几十 MB），YARN 会直接在 ApplicationMaster 所在的 Container 里串行执行所有 MapTask 和 ReduceTask，不需要额外申请 Container。

启用条件（默认关闭）：

<property>
  <name>mapreduce.job.ubertask.enable</name>
  <value>true</value>
</property>

Uber 模式减少了小 Job 的调度开销（不需要为几个秒级的任务申请、释放 Container），但只适合 map 数 ≤ 3、reduce 数 ≤ 1、输入数据量很小的场景。

3.15 MapReduce 调优踩坑合集

1. OOM（Out Of Memory）

MapTask 或 ReduceTask 频繁被 YARN kill（报错 Container killed by the ApplicationMaster 或 Exit code: 143），大概率是内存不够。

<!-- mapred-site.xml -->
<property>
  <name>mapreduce.map.memory.mb</name>
  <value>4096</value>   <!-- 默认 1024，不够就往上加 -->
</property>
<property>
  <name>mapreduce.reduce.memory.mb</name>
  <value>8192</value>
</property>
<property>
  <name>mapreduce.map.java.opts</name>
  <value>-Xmx3276m</value>  <!-- JVM 堆内存，通常是 container 内存的 0.8 倍 -->
</property>

2. JVM 重用（JVM Reuse）

默认每个 Task 都启动一个新 JVM，启动开销几百毫秒到几秒不等。如果 Job 里有成千上万个短 Task，JVM 启动时间可能比实际计算时间还长。

<property>
  <name>mapreduce.job.jvm.numtasks</name>
  <value>10</value>  <!-- 一个 JVM 串行执行 10 个 Task -->
</property>

注意：JVM 重用后，静态变量不会自动清空，如果你在 Mapper 里用了静态变量缓存数据，下一个 Task 可能会读到脏数据。

3. 数据倾斜

ReduceTask 之间运行时间差距极大（有的几秒，有的几小时），就是典型的数据倾斜。根因通常是某个 key 的数据量远大于其他 key（比如空值 key、热点用户）。

解决方案：

• 自定义 Partitioner，把热点 key 打散到多个 ReduceTask
• 在 Mapper 端对热点 key 加随机前缀，Reduce 后再二次聚合
• 如果倾斜 key 已知，可以单独为它开一个 Job 处理

4. 小文件问题

MapReduce 的每个文件至少对应一个 MapTask（除非用 CombineTextInputFormat 合并小文件）。如果输入目录里有 10 万个 1KB 的小文件，会启动 10 万个 MapTask，调度开销爆炸。

解决方案：

• 用 CombineTextInputFormat 合并小文件
• 写数据时用 SequenceFile 或 Avro 打包
• 跑个合并 Job，把历史小文件合并成大文件
• HDFS 层面用 Ozone 替代（小文件友好）

---

四、YARN 资源调度平台

4.1 YARN 解决了什么问题

YARN 出来之前，Hadoop 1.x 时代 MapReduce 和资源管理是绑在一起的，集群只能跑 MapReduce，Spark、Flink 这些框架根本上不去。

YARN 的核心设计是解耦：把资源管理（ResourceManager + NodeManager）和计算框架（ApplicationMaster + Container）分开。集群资源由 YARN 统一调度，各种计算框架以 Application 的形式接入。

4.2 YARN 架构

ResourceManager（RM）

整个集群的资源大脑，负责：

• 接收 Client 提交的 Application
• 与 NodeManager 保持心跳，掌握全局资源状况
• 通过 Scheduler 把资源分配给各个 Application

RM 内部有两个核心组件：

• Scheduler：纯调度器，只负责资源分配，不管任务执行状态。支持 FIFO、Capacity、Fair 三种调度策略
• ApplicationsManager：负责管理 ApplicationMaster 的生命周期（启动、监控、重启）

NodeManager（NM）

每个节点上都有一个 NM，负责：

• 向 RM 定期发送心跳，汇报本节点的资源使用情况
• 接收 RM 的指令，在本地启动/停止 Container
• 监控本节点上所有 Container 的资源使用，杀死超资源的 Container

ApplicationMaster（AM）

每个 Application 都有一个 AM，是"应用级别"的资源管家：

• 向 RM 申请本应用所需的 Container 资源
• 与 NM 通信，让 NM 在指定 Container 里启动任务
• 监控任务的执行状态，失败的任务重新申请资源重跑
• 应用完成后向 RM 注销自己

MapReduce 的 AM 实现是 MRAppMaster，Spark 的 AM 实现是 Spark Driver（在 cluster 模式下）。

Container

YARN 的资源抽象单位，包含：

• 内存（默认最低 1GB，yarn.scheduler.minimum-allocation-mb）
• CPU 核数（默认 1 核，yarn.scheduler.minimum-allocation-vcores）
• 磁盘、网络等资源（Hadoop 3.x 支持 GPU 等扩展资源）

Container 由 RM 分配，NM 启动和监控，AM 使用。

4.3 调度器对比

YARN 支持三种调度器：

调度器	特点	适用场景
FIFO	先进先出，单队列	测试环境、小规模集群
Capacity	多队列，预设容量比例，弹性共享	多部门共享集群，资源隔离
Fair	多队列，按权重公平分配，支持抢占	需要严格公平性的多租户场景

生产环境最常用的是 Capacity Scheduler。配置示例：

<!-- capacity-scheduler.xml -->
<property>
  <name>yarn.scheduler.capacity.root.queues</name>
  <value>prod,dev,test</value>
</property>
<property>
  <name>yarn.scheduler.capacity.root.prod.capacity</name>
  <value>60</value>
</property>
<property>
  <name>yarn.scheduler.capacity.root.dev.capacity</name>
  <value>30</value>
</property>
<property>
  <name>yarn.scheduler.capacity.root.test.capacity</name>
  <value>10</value>
</property>

4.4 Hadoop 3.x YARN 新特性

节点标签（Node Labels）

允许给节点打标签，把集群逻辑上划分成多个分区。比如：

• 给 GPU 机器打标签 GPU
• 给高内存机器打标签 HIGH_MEM
• 给普通机器打标签 DEFAULT

提交任务时可以指定用哪个标签的节点：

# 给节点加标签
yarn rmadmin -addToClusterNodeLabels "GPU(exclusive)"
yarn rmadmin -replaceLabelsOnNode "node1:GPU"

# 提交任务到 GPU 节点
yarn jar app.jar -Dmapreduce.job.node-label-expression=GPU

GPU 调度与隔离

Hadoop 3.x 原生支持 NVIDIA GPU 的调度：

<!-- resource-types.xml -->
<property>
  <name>yarn.resource-types</name>
  <value>yarn.io/gpu</value>
</property>

需要配 DominantResourceCalculator，NM 会自动通过 nvidia-smi 探测 GPU 设备，任务提交时可以申请 GPU 资源。配合 Docker 运行时，可以实现 GPU 的容器级隔离。

Opportunistic Containers（机会型容器）

传统 YARN 的 Container 是 Guaranteed 的：一旦分配给你，就必须立即有资源执行。

Hadoop 3.x 引入了 Opportunistic Container：即使没有空闲资源也可以先排队，等别的任务释放资源后再执行。优先级低于 Guaranteed Container，可以被抢占。

这个机制能提高集群的整体利用率，特别适合短任务、低优先级任务的场景。

<property>
  <name>yarn.resourcemanager.opportunistic-container-allocation.enabled</name>
  <value>true</value>
</property>

Docker 与 K8s 集成

Hadoop 3.x 支持 Docker 作为 Container 的运行时环境，任务可以跑在隔离的 Docker 容器里。同时 YARN 也支持作为 K8s 的底层调度层，或者通过 YARN Federation 实现跨集群调度。

4.5 ResourceManager HA

ResourceManager 也有单点故障风险，Hadoop 2.4+ 引入了 RM HA 机制。

和 NameNode HA 类似，RM HA 也采用 Active/Standby 模式：

• Active RM：处理所有客户端请求，调度资源
• Standby RM：同步 Active 的状态，随时准备接管

状态持久化方式有两种：

• 基于 ZooKeeper：通过 ZK 做 leader 选举和状态存储（生产环境推荐）
• 基于文件系统：把状态写到共享文件系统（如 HDFS），Standby 轮询读取

自动切换机制：ZKFC 监控 Active RM 的心跳，如果挂了，ZKFC 触发 Standby 提升为 Active。正在运行的 Application 不受影响（ApplicationMaster 会缓存 RM 地址列表，自动重连新的 Active）。

<!-- yarn-site.xml -->
<property>
  <name>yarn.resourcemanager.ha.enabled</name>
  <value>true</value>
</property>
<property>
  <name>yarn.resourcemanager.ha.rm-ids</name>
  <value>rm1,rm2</value>
</property>

4.6 资源隔离：CGroups

YARN 默认只做逻辑隔离（分配你 4GB 内存，你申请了就扣掉，但不阻止你实际使用 8GB）。这在多租户场景下很坑：某个 Bug 程序内存泄漏，会把整台机器的内存吃光，导致其他 Container 被 OOM kill。

从 Hadoop 2.x 后期版本开始，一直到 3.x，都支持 CGroups 物理隔离：

<!-- yarn-site.xml -->
<property>
  <name>yarn.nodemanager.linux-container-executor.cgroups.mount</name>
  <value>true</value>
</property>
<property>
  <name>yarn.nodemanager.container-executor.class</name>
  <value>org.apache.hadoop.yarn.server.nodemanager.LinuxContainerExecutor</value>
</property>

启用后，YARN 会为每个 Container 创建独立的 CGroup，限制 CPU 和内存使用。Container 想超资源？直接 OOM 或者 CPU throttle，不会影响其他任务。

踩坑点：LinuxContainerExecutor 权限

LinuxContainerExecutor 需要配置 container-executor 二进制文件的 setuid 权限（属主 root，权限 6050），否则无法创建 CGroup。很多自己编译 Hadoop 的人在这里卡住，因为官方 tarball 里的 container-executor 默认是普通权限，需要手动 chmod 6050 和 chown root:hadoop。

4.7 YARN 调优踩坑合集

1. Container 虚拟核数 vs 物理核数

yarn.nodemanager.resource.cpu-vcores 配置的是虚拟核数，不是物理核数。比如一台 32 核的机器，你配 32 vcores，然后每个 Container 申请 4 vcores，同时跑 8 个 Container，这时候是 8 个进程抢 32 个物理核。但如果 YARN 不知道你机器的真实物理核数（比如通过 CGroups），CPU 会严重超卖，任务之间互相抢占，导致性能抖动。

建议：vcores 的数量按物理核数的 1.5 ~ 2 倍设置（留一些超线程余量），同时启用 CGroups 做硬隔离。

2. 内存配置不匹配

YARN 的 Container 内存和 JVM 堆内存是两个概念，很多人混淆：

• mapreduce.map.memory.mb = 4096：Container 申请 4GB
• mapreduce.map.java.opts = -Xmx3276m：JVM 堆只分配 3.2GB

中间差的 800MB 是给非堆内存（Native 内存、线程栈、Direct Buffer）留的。如果你把 java.opts 也设成 4096m，Container 实际使用会超过 4GB，被 NM kill。

黄金法则：java.opts = 0.8 * container.memory.mb。

3. AM 日志找不到

Application 跑完后，AM 的日志默认在 NodeManager 的本地目录（yarn.nodemanager.log-dirs），不是 HDFS。如果 NodeManager 重启了或者磁盘清了，日志就没了。

生产环境必须开 Log Aggregation：

<property>
  <name>yarn.log-aggregation-enable</name>
  <value>true</value>
</property>
<property>
  <name>yarn.nodemanager.remote-app-log-dir</name>
  <value>/tmp/logs</value>
</property>

开启后，Application 完成后 NM 会自动把本地日志聚合到 HDFS 上，YARN UI 可以直接查看历史日志。

4. 队列资源饿死

Capacity Scheduler 的队列有 minimum-user-limit-percent 和 user-limit-factor 两个参数，默认值可能导致大 Job 占满队列后小 Job 永远进不来。

<!-- 每个用户最多占用队列资源的 50%，保证其他用户有资源 -->
<property>
  <name>yarn.scheduler.capacity.root.prod.minimum-user-limit-percent</name>
  <value>50</value>
</property>

如果不限制，某个用户提交了一个占满整个队列的 Job，其他用户的 Job 会一直 pending。

---

五、Hadoop 3.x 新特性总览

除了前面分章节提到的，这里再集中梳理一下 Hadoop 3.x 的核心特性：

5.1 HDFS 层

• 纠删码（Erasure Coding）：存储开销砍半，适合冷数据归档
• 多 NameNode HA：支持 1 Active + N Standby，不再局限于 1+1
• HDFS 磁盘均衡器（Disk Balancer）：Intra-node 级别做磁盘间数据平衡（之前只有 Inter-node 的 Balancer）
• Intra-DataNode 均衡：单节点多块磁盘之间的数据均衡
• S3A 增强：更好的云存储集成，支持对象存储

5.2 YARN 层

• Timeline Service v2：基于 HBase 存储，支持更细粒度的应用历史监控和查询
• Opportunistic Containers：提高短任务集群利用率
• 分布式调度：NodeManager 本地调度器处理机会型容器，减少 RM 压力
• 节点标签：细粒度控制任务跑在哪些机器上
• GPU/FPGA 资源调度：支持异构硬件调度
• Docker 容器化运行时：任务隔离更安全

5.3 MapReduce 层

• Native Map Output Collector：Shuffle 密集型任务性能提升 30%+
• Task 级 Native 优化：基于 JNI 的排序和序列化加速

5.4 生态新成员

• Apache Ozone：分布式对象存储，弥补 HDFS 小文件缺陷，原生支持 S3 API
• Apache Submarine（已独立成 Apache 顶级项目）：深度学习工作流管理平台，支持 TensorFlow/PyTorch 在 YARN 上跑训练任务

---

六、总结

Hadoop 这套体系虽然已经有近 20 年历史，但它的核心组件（HDFS + YARN）至今仍是很多大数据平台的底座。MapReduce 确实被 Spark、Flink 这些新一代计算引擎取代了很多场景，但理解 MapReduce 的分治思想和 Shuffle 机制，对理解整个分布式计算的底层原理仍然很有帮助。

Hadoop 3.x 的升级重点在于存储效率和资源调度的现代化：纠删码省了一半存储成本，YARN 支持 GPU 和容器化让 Hadoop 集群能跑 AI 训练任务，Ozone 补上了小文件和对象存储的短板。如果你手上有老集群，升级到 3.x 的性价比还是很高的。

最后提一句：现在云厂商的对象存储（OSS/S3）和 Serverless 计算越来越便宜，不少公司已经把 Hadoop 集群迁移到云上或者用 EMR 这类托管服务。自建 Hadoop 集群的运维成本确实不低，选型时还要把人力成本算进去。