hadoop基础知识分享(二)
写在前面今天继续学习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 作业，可以显著提高作业的执行效率。
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
页:
[1]