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 方法来指定排序规则。
自定义排序示例：

1. public class MyCustomComparator implements WritableComparable<MyCustomComparator> {
2.     private int key;
3.    
4.     @Override
5.     public int compareTo(MyCustomComparator other) {
6.         // 比较 key 的大小
7.         return Integer.compare(this.key, other.key);
8.     }
9.    
10.     // 重写其他必要的序列化和反序列化方法
11. }

复制代码

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 作业，可以显著提高作业的执行效率。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！