写在前面
本日继续学习hadoop部分的知识。
MapReduce
数据切片
InputFormat 接口
InputFormat 是一个抽象类,界说了两个方法:
- getSplits:负责实现数据切片的方法
- createRecordReader:实现数据的 key-value 格式。
FileInputFormat 抽象类
FileInputFormat 是所有以文件作为数据源的 InputFormat 实现的基类,保存作为 job 输入的所有文件,并实现了对输入文件盘算 splits 的方法。不同的子类(如 TextInputFormat)负责获得记录的方法。
TextInputFormat
TextInputFormat 是默认的处理类,用于处理平凡文本文件。文件中的每一行作为一个记录:
- Key:每一行的起始偏移量
- Value:每一行的内容
默认分隔符是换行符(\n)或回车键。
KeyValueTextInputFormat
适用于以制表符(\t)分割的多列数据:
- Key:分隔符前的部分
- Value:分隔符后的部分
如果没有分隔符,整行作为 Key,Value 为空。
SequenceFile 输入格式
适用于二进制文件。
CombineFile 输入格式
适用于小文件,解决大量小文件导致的 MapTask 数量过多问题。
InputSplit的巨细
InputSplit 的巨细由以下公式决定:
Math.max(minSize, Math.min(maxSize, blockSize))
- 单节点建议运行 10—100 个 Map Task。
- Map Task 实行时长不建议低于 1 分钟,否则效率低。
特别:一个输入文件巨细为 140M,应该有几个 Map Task?
- 由 FileInputFormat 类中的 getSplits 方法决定,通常为 1 个 Map Task。
实行流程
Map 任务处理
- 框架使用 InputFormat 的子类把输入文件(夹)划分为多个 InputSplit,默认每个 HDFS 块对应一个 InputSplit。
- RecordReader 将每个 InputSplit 解析成 。
- 调用 Mapper 类中的 map(...) 函数,输入是 ,输出是 。
- 如果有 Reduce,框架会对 Map 输出的 进行分区,默认只有 1 个分区。
- 数据按 k2 进行排序和分组,相同 k2 的 v2 会分为一个组。
Reduce 任务处理
- 在 Reduce 阶段,框架会将多个 Map 任务的输出通过网络传输到不同的 Reduce 节点,这个过程称为 Shuffle。
- 框架对接收到的相同分区的 数据进行合并、排序、分组。
- 调用 Reducer 类中的 reduce 方法,输入为 ,输出为 。
- 最终结果保存到 HDFS 中。
自界说输出类
- FileOutputFormat 是默认的输出类。如果需要自界说输出格式,需要继承 FileOutputFormat 并在 RecordWriter 中根据自界说的输出逻辑重写方法。
YARN
资源管理框架
架构
YARN 是 Hadoop 的资源管理框架,接纳主从架构:
- ResourceManager:管理集群资源。
- NodeManager:管理每个节点的资源(如 CPU 和内存)。
- ApplicationMaster:任务的调度和资源请求,需要实现 ApplicationMaster 接口才能提交到 YARN 上。
调度计谋
- FIFO:先进先出调度计谋。
- 公平调度:每个作业按比例分配资源。
- 容量调度:为每个作业分配固定的资源容量。
Shuffle 过程
Shuffle 过程简介
广义的 Shuffle 过程是指,在 Map 函数输出数据之后,并且在 Reduce 函数实行之前的过程。Shuffle 过程包含了数据的分区、溢写、排序、合并等操纵。具体而言,Shuffle 主要包括以下步调:
- 分区(Partitioning):根据自界说的分区规则,将 Map 输出的 数据分配到不同的 Reduce 任务。
- 溢写(Spill):当内存中保存的 Map 输出数据超过阈值时,数据会被溢写到本地磁盘。
- 排序(Sorting):Map 输出的数据会按照 Key 值进行排序,默认接纳字典顺序排序。
- 合并(Merging):多个 Map 任务的输出会被合并,确保每个分区的数据都按顺序分列并预备好交给 Reducer 进行处理。
Shuffle 源码实现
Shuffle 过程的核心代码主要包含在 MapOutputCollector 的子实现类中。该类对象表示缓冲区,负责存储 Map 输出数据并处理 Shuffle 阶段的操纵。
自界说排序
Shuffle 的默认排序方式是按照 Hash 值的字典顺序进行排序。如果需要自界说排序,可以通过实现 WritableComparable 接口并重写 compareTo 方法来指定排序规则。
自界说排序示例:- public class MyCustomComparator implements WritableComparable<MyCustomComparator> {
- private int key;
-
- @Override
- public int compareTo(MyCustomComparator other) {
- // 比较 key 的大小
- return Integer.compare(this.key, other.key);
- }
-
- // 重写其他必要的序列化和反序列化方法
- }
复制代码 InputFormat、FileInputFormat、TextInputFormat的区别?
这些类主要用于 Map 阶段的数据输入、切片和格式转换。
- InputFormat 接口:
InputFormat 是一个接口,界说了两个方法:
- getSplits:负责实现数据切片。
- createRecordReader:将每个切片中的数据转换为 key-value 格式。
- FileInputFormat:
FileInputFormat 是一个实现了 getSplits 方法的抽象基类,负责将输入数据切片,并处理数据为 key-value 格式。它实现了对数据切片的功能,默认按照文件的 HDFS 块进行切分,且溢出块最多为 10%。
- TextInputFormat:
TextInputFormat 是 FileInputFormat 的具体实现类,用于处理文本文件。它将每一行的偏移量作为 key,每一行的内容作为 value,并默认使用换行符作为行分隔符。它是 MapReduce 的默认输入格式。
MapReduce 实行流程?
- Map 阶段:
- 数据切片:Map 阶段起首将输入数据切片,每个切片对应一个 Map Task。默认情况下,TextInputFormat 会将数据按照 HDFS 块切分。
- 格式化数据:每个切片的数据会通过 RecordReader 转换为 key-value 对,默认情况下,偏移量为 key,每行文本内容为 value。
- Map 输出:Map Task 处理每个输入的 key-value 对后,天生新的输出 k2-v2 对。输出会先进入一个 100MB 巨细的环形缓冲区。当缓冲区中的数据达到 80%(80MB)时,数据会被溢写到磁盘。
- 分区和排序:Map 输出的 k2-v2 对会进行分区,分区的数量与 Reduce 任务的数量相同。分区后,数据会按 key 排序(默认是字典顺序)。
- Combiner 归约(可选):如果设置了 Combiner,Map 输出会在写入磁盘进步行局部聚合。
- Shuffle 阶段:
- 数据传输:Map 输出的每个分区会通过网络传输到相应的 Reduce 节点,这个过程叫做 Shuffle。
- 数据合并和排序:在 Reduce 端,接收到的数据会按照 key 进行排序和分组。相同 key 的 value 会被合并为一个集合。
- Reduce 阶段:
- 数据处理:Reduce Task 接收到 key 和 value 集合后,调用 reduce 方法对数据进行聚合。不同的 key 会进入不同的 Reduce Task 中进行处理。
- 结果输出:Reduce Task 将最终的输出写入 HDFS。
如何在 Map 阶段进行 Reduce 操纵(MapJoin)?
使用 Combiner 可以在 Map 阶段进行局部的聚合,从而淘汰 Reduce 阶段的数据量。Combiner 是一个在 Map 完成后实行的“轻量级” Reduce 操纵,它会对每个 Map Task 输出的数据进行初步的归约,淘汰 Shuffle 阶段的数据传输量。
- MapJoin:在 setup 方法中使用小表数据与大表数据进行连接,制止 Reduce 阶段的复杂盘算。
MapReduce 优化?
数据输入优化
合并小文件
问题:大量小文件会导致创建大量 Map Task,从而低落实行效率。
解决方案:
- 使用 CombineTextInputFormat 代替默认的 TextInputFormat,制止为每个小文件创建单独的 Map Task。
- 在读取数据之前,先合并小文件,淘汰小文件导致的 Map Task 数量。
Map 阶段优化
淘汰溢写次数
问题:多次溢写会导致磁盘 I/O 开销大,低落效率。
解决方案:
- 增大溢写触发的内存上限,淘汰溢写次数:
- 设置 io.sort.mb(环形缓存区巨细)来增大内存缓冲区。
- 设置 sort.spill.percent 来调解溢写触发的比例(默以为 80%)。
- 使用 Combine 进行预聚合,淘汰 I/O。
- 使用 MapJoin 优化数据合并操纵,在 Map 阶段完成预处理。
I/O 传输优化
问题:大量数据传输会影响 MapReduce 实行效率。
解决方案:
- 接纳数据压缩淘汰网络 I/O 时间。
- 使用 Combine 或提前过滤数据来淘汰传输的数据量。
- 适当使用本地化 Map 任务,淘汰跨节点的数据传输。
Reduce 阶段优化
问题: Reduce 阶段实行效率慢。
解决方案:
- 合理设置 Reduce 任务的数量,制止过多的 Reduce Task。
- 使用 MapJoin 来淘汰 Reduce 阶段的数据量。
- 使数据分布匀称,制止数据倾斜。
MapReduce 在 YARN 上的实行流程?
- 资源申请:
- ResourceManager 会为每个应用(如 MapReduce 作业)在 NodeManager 中申请资源容器(container)。每个容器会分配肯定量的内存、CPU 和硬盘资源,并启动一个 JVM 进程。
- ApplicationMaster 会为作业中的每个 Task(Map 或 Reduce)向 ResourceManager 申请容器,并请求启动该容器。
- Task 实行:
- NodeManager 启动并分配容器资源后,实行相应的 Map 或 Reduce Task。
- Task 实行完后,容器资源会被开释,NodeManager 收回容器。
粗粒度与细粒度资源申请
- 细粒度资源申请(MapReduce):每个任务根据需要动态申请资源,比方 1GB 内存,只使用所需的资源。
- 粗粒度资源申请(Spark):任务会申请一个固定巨细的资源,不管任务巨细如何,直到任务完成才开释资源。Spark 需要更多的内存资源,适用于内存密集型盘算。
YARN 的调度计谋?
YARN 中有三种调度计谋,分别适用于不同的场景:
- FIFO Scheduler(先进先出调度器):
按照作业提交的顺序进行调度。简单易用,但可能导致其他作业的资源请求被阻塞,不推荐在生产环境中使用。
- Capacity Scheduler(容量调度器):
为不同的队列分配固定的资源容量。适用于多租户集群,每个部门或项目都能获得肯定的资源份额。
- Fair Scheduler(公平调度器):
每个作业根据公平原则动态调解资源分配,适用于多个作业并发实行,包管每个作业公平地分配资源。CDH 中的默认调度计谋。
通过合理选择调度计谋和优化 MapReduce 作业,可以明显进步作业的实行效率。
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。 |