Hadoop 学习总结

Posted 2022-09-12 jianminglin

本文由我手动整理自 W3Cschool Hadoop 教程 (https://www.w3cschool.cn/hadoop/)，看不懂就手敲了一遍，好烦呀

Hadoop 关于

大数据概念

• 不能使用一台机器进行处理数据
• 大数据的核心是样本=总体

大数据特性

• 大量性(volume): 一般在大数据里，单个文件的级别至少为几十，几百GB以上
• 快速性(velocity): 反映在数据的快速产生及数据变更的频率上
• 多样性(variety): 泛指数据类型及其来源的多样化，进一步可以把数据结构归纳为结构化(structured)，半结构化(semi-structured)，和非结构化(unstructured)
• 易变性: 伴随数据快速性的特征，数据流还呈现一种波动的特征。不稳定的数据流会随着日，季节，特定事件的触发出现周期性峰值
• 准确性: 又称为数据保证(data assurance)。不同方式，渠道收集到的数据在质量上会有很大差异。数据分析和输出结果的错误程度和可信度在很大程度上取决于收集到的数据质量的高低
• 复杂性: 体现在数据的管理和操作上。如何抽取，转换，加载，连接，关联以把握数据内蕴的有用信息已经变得越来越有挑战性

关键技术

1. 数据分布在多台机器上

   • 可靠性：每个数据块都复制到多个节点
   • 性能：多个节点同时处理数据

2. 计算随数据走

   • 网络IO速度 << 本地磁盘 IO 速度，大数据系统会尽量地将任务分配到离数据最近的机器上运行（程序运行时，将程序及其依赖包都复制到数据所在的机器运行）
   • 代码向数据迁移，避免大规模数据时，造成大量数据迁移的情况，尽量让一段数据的计算发生在同一台机器上

3. 串行 IO 取代随机 IO

   传输时间 << 寻道时间，一般数据写入后不在修改

Hadoop 简介

概念

Hadoop 可运行与一般的商用机器上，具有高容错，高可靠性，高扩展等特点

特别适合写一次，读多次的场景

适用场景

• 大规模数据
• 流式数据（写一次，读多次）
• 商用硬件（一般硬件）

不适用场景

• 低延时的数据访问
• 大量的小文件
• 频繁修改文件（基本就是写1次）

Hadoop 架构

• HDFS：分布式文件存储
• YARN：分布式资源管理
• MapReduce：分布式计算
• Others：利用YARN的资源管理功能实现其他的数据处理方式

内部各个节点基本都是采用 Master-Worker 架构

Hadoop HDFS

Hadoop Distributed File System，分布式文件系统

HDFS 架构

• Block 数据块

  1. 基本存储单位，一般大小为 64M，配置大的块主要因为：
     • 减少搜索时间，一般硬盘传输速率比寻道时间要快，大的块可以减少寻道时间；
     • 减少管理块的数据开销，每个块都需要在 NameNode 上有对应的记录；
     • 对数据块进行读写，减少建立网络的连接成本。
  2. 一个大文件会被拆分为一个个的块，然后存储于不同的机器上。如果一个文件小于 Block 大小，那么实际占用空间为其文件的大小。
  3. 基本的读写单位，类似磁盘的页，每次都是读写一个块。
  4. 每个块都会被复制到多台机器，默认复制3份。

• NameNode

  1. 存储文件的 metadata，运行时所有数据都保存到内存，整个HDFS可存储的文件数受限于 NameNode 的内存大小。
  2. 一个Block在 NameNode 中对应一条记录（一般一个Block占用150字节），如果是大量的小文件，会消耗大量内存。同时 map task 的数量使用 splits 来决定的，所以用 MapReduce 处理大量的小文件时，就会产生过多的 map task，线程管理开销将会增加作业时间。处理大量小文件的速度远远小于处理同等大小的大文件的速度。因此 Hadoop 建议存储大文件。
  3. 数据会定时保存到本地磁盘，但不保存 Block 的位置信息，而是由 DataNode 注册时上报和运行时维护（NameNode 中与 DataNode 相关的信息并不保存到 NameNode 的文件系统中，而是 NameNode 每次重启后，动态创建）。
  4. NameNode 失效则整个HDFS都失效了，所以要保证 NameNode 的可用性。

• Secondary NameNode

  定时与 NameNode 进行同步（定期合并文件系统镜像和编辑日志，然后把合并后的结果传给 NameNode，替换其镜像，并清空编辑日志，类似于 CheckPoint 机制），但 NameNode 失效后仍需要手工将其设置成主机。

• DataNode

  1. 保存具体的 Block 数据。
  2. 负责数据的读写操作和复制操作。
  3. DataNode 启动时会向 NameNode 报告当前存储的数据块信息，后续也会定时报告修改信息。
  4. DataNode 之间会进行通信，复制数据块，保证数据的冗余性。

HDFS 写文件

1. 客户端将文件写入本地磁盘的文件中

2. 当临时文件大小达到一个Block大小时，HDFS Client 通知 NameNode，申请写入文件

3. NameNode 在HDFS的文件系统中创建一个文件，并把该 Block ID 和要写入的 DataNode 的列表返回给客户端

4. 客户端收到这些消息后，将临时文件写入 DataNodes

   1. 客户端将文件内容写入第一个 DataNode（一般以 4kb 单位进行传输）。
   2. 第一个 DataNode 接收后，将数据写入本地磁盘，同时也传输给第二个 DataNode。
   3. 以此类推到最后一个 DataNode，数据在 DataNode 之间是通过 pipeline 的方式进行复制的。
   4. 后面的 DataNode 接受完数据后，都会发送一个确认给前一个 DataNode，最终第一个 DataNode 返回确认给客户端。
   5. 当客户端接收到整个 Block 的确认后，会向 NameNode 发送一个最终的确认信息。
   6. 如果写入某个 DataNode 失败，数据会继续写入其他的 DataNode。然后 NameNode 会找另一个好的 DataNode 继续复制，以保证冗余性。
   7. 每个Block 都会有一个校验码，并存放到独立的文件中，以便读的时候验证其完整性。

5. 文件写完后（客户端关闭），NameNode 提交文件（这时文件才可见，如果提交前，NameNode 挂掉，那文件也就丢失了。

   fsync：只保证数据的信息写到 NameNode 上，但并不保证数据已经被写道 DataNode 中）

Rack Aware（机架感知）

• 通过配置文件制定机架名和DNS的对应关系

• 假设复制参数是3，在写入文件时，会在本地的机架保存一份数据，然后再另一个机架内保存两份数据（同机架内的传输速度快，从而提高性能）

• 整个HDFS的集群，最好是负载均衡的，这样才能尽量利用集群的优势

HDFS 读文件

1. 客户端向 NameNode 发送读取请求。
2. NameNode 返回文件的所有 Block 和这些 Block 所在的 DataNodes（包括复制节点）。
3. 客户端直接从 DataNode 中读取数据，如果该 DataNode 读取失败（DataNode 失效或校验码不对），则从复制节点中读取（如果读取的数据就在本机，则直接读取，否则通过网络读取）。

HDFS 可靠性

1. DataNode 可以失效

   DataNode 会定时发送心跳到 NameNode。如果在一段时间内 NameNode 没有收到 DataNode 的心跳信息，则认为其失效。此时 NameNode 就会将该节点的数据（从该节点的复制节点中获取）复制到另外的 DataNode 中。

2. 数据可以损坏

   无论是写入时还是硬盘本身的问题，只要数据有问题（读取时通过校验码来检测），都可以通过其他的复制节点读取，同时还会再复制一份到健康的节点中。

3. NameNode 不可靠

HDFS 命令工具

• fsck：检查文件的完整性
• start-balancer.sh：重新平衡 HDFS
• hdfs dfs -copyFromLocal：从本地磁盘复制文件到 HDFS

Hadoop YARN

旧版架构

• JobTracker：负责资源管理，跟踪资源消耗和可用性，作业生命周期管理（调度作业任务，跟踪进度，为任务提供容错）
• TaskTracker：加载或关闭任务，定时报告任务状态

架构存在的问题：

1. JobTracker 是 MapReduce 的集中式处理点，存在单点故障。
2. JobTracker 完成了太多的任务，造成了过多的资源消耗，当 MapReduce Job 非常多的时候，会造成很大的内存开销。这也是业界普遍总结出来 Hadoop 的 MapReduce 只能支持 4000 节点主机的上限。
3. 在 TaskTracker 端，以 map/reduce task 的数目作为资源的表示过于简单，没有考虑到 cpu/内存 的占用情况。如果两个大内存消耗的 task 被调度到一块，很容易出现 OOM。
4. 在 TaskTracker 端，把资源强制划分为 Map task slot 和 Reduce task slot，如果当系统中只有 Map task 或者只有 Reduce task 的时候，会造成资源的浪费，也就是集群资源利用的问题。

总的来说，就是 单点问题 和 资源利用率问题。

新版架构 YARN

YARN 就是将 JobTracker 的职责进行拆分，将资源管理和任务调度拆分成独立的进程：一个全局的资源管理（ResourceManager）和一个单个作业的管理（ApplicationMaster）。ResourceManager 和 NodeManager 提供计算资源的分配和管理，为 ApplicationMaster 完成应用程序的运行。

• ResourceManager：全局资源管理和任务调度
• NodeManager：单个节点的资源管理和监控
• ApplicationMaster：单个作业的资源管理和任务监控
• Container：资源申请的单位和任务运行的容器

新旧架构对比

YARN 架构下形成了一个通用的资源管理平台和一个通用的应用计算平台，避免了旧架构的单点问题和资源利用率问题，同时也让在其上在运行的应用不再局限于MapReduce形式。

YARN 基本流程

1. Job Submission

   从 ResourceManager 中获取一个 Application ID 检查作业输出配置，计算输入分片、拷贝作业资源（Job jar、配置文件、分片信息）到 HDFS，以便后面任务的执行。

2. Job Initialization

   • ResourceManager 将作业递交给 Scheduler（有很多调度算法，一般是根据优先级）。

   • Scheduler为作业分配一个 Container，ResourceManager 就加载一个 application master process 并交给 NodeManager 管理。

   • ApplicationMaster 主要是创建一系列的监控进程来跟踪作业的进度，同时获取输入分片，为每一个分片创建一个Map task 和响应的 Reduce task。

     ApplicationMaster 还决定如何运行作业，如果作业很小（可配置），则直接在同一个 JVM 下运行。

3. Task Assignment

   ApplicationMaster 向 ResourceManager 申请资源（一个个的 Container，指定任务分配的资源要求），一般是根据 data locality 来分配资源。

4. Task Execution

   ApplicationMaster 根据 ResourceManager 的分配情况，在对应的 NodeManager 中启动 Container，从HDFS中读取任务所需的资源（job jar，配置文件等），然后执行该任务。

5. Progress and Status Update

   定时将任务的进度和状态报告给 ApplicationMaster Client，定时向 ApplicationMaster 获取整个任务的进度和状态。

6. Job Completion

   Client 定时检查整个作业是否完成。作业完成后，会清空临时文件、目录等。

YARN ResourceManager

负责全局的资源管理和任务调度，把整个集群当成计算资源池，只关注分配，不管应用，且不负责容错。

资源管理

1. 以前资源是每个节点分成一个个的 Map slot 和 Reduce slot，现在是一个个 Container，每个 Container 可以根据需要运行 ApplicationMaster、Map、Reduce 或者任意程序
2. 以前资源分配是静态的，目的是动态的，资源利用率更高
3. Container 是资源申请的单位，一个资源申请格式：<resource-name,priority,resource-requirement,number-of-containers>
   • resource-name:主机名、机架名或*（代表任意机器）
   • resource-requirement：目前只支持CPU和内存
4. 用户提交作业到 ResourceManager，然后再某个 NodeManager 上分配一个 Container 来运行 ApplicationMaster，ApplicationMaster 再根据自身程序需要向 ResourceManager 申请资源。
5. YARN有一套 Container 的生命周期管理机制，而 ApplicationMaster 和其 Container 之间的管理是应用程序自己定义的。

任务调度

1. 只关注资源的使用情况，根据需求合理分配资源。
2. Scheduler 可以考虑申请的需要，在特定的机器上申请特定的资源（ApplicationMaster 负责申请资源时的数据本地化的考虑，ResourceManager将尽量满足其申请需求，在制定的机器上分配 Container，从而减少数据移动）。

内部结构

• Client Service：应用提交、终止、输出信息（应用、队列、集群等的状态信息）。
• Admin Service：队列、节点、Client权限管理。
• ApplicationMasterService：注册、终止ApplicationMaster，获取ApplicationMaster 的资源申请或取消的请求，并将其异步地传给 Scheduler，单线程处理。
• ApplicationMaster Liveliness Monitor：接收ApplicationMaster 的心跳消息，如果某个 ApplicationMaster 在一定时间内没有发送心跳，则被认为任务失败，其资源将会被回收，然后 ResourceManager 会重新分配一个 ApplicationMaster 运行该应用（默认尝试 2次）。
• Resource Tracker Service：注册节点，接收各注册节点的心跳消息。
• NodeManagers Liveliness Monitor：监控每个节点的心跳消息，如果长时间没有接收到心跳消息，则认为该节点无效，同时所有在该节点上的 Container 都标记成无效，也不会调度任务到该节点运行。
• ApplicationManager：管理应用程序，记录和管理已完成的应用。
• ApplicationMaster Launcher：一个应用提交后，负责与 NodeManager 交互，分配 Container 并加载 ApplicationMaster，也负责终止或销毁。
• YarnScheduler：资源调度分配，有 FIFO，Fair，Capacity 方式
• ContainerAllocationExpirer：管理已分配但没有启用的 Container，超过一定时间则将其回收。

YARN NodeManager

Container 管理

1. 启动时向 ResourceManager 注册并定时发送心跳信息，等待 ResourceManager的指令。
2. 监控 Container 的运行，维护 Container 的生命周期，监控 Container 的资源使用情况。
3. 启动或停止 Container，管理任务运行时的依赖包（根据 ApplicationMaster 的需要，启动 Container 之前将需要的程序机器依赖包、配置文件等拷贝到本地）。

内部结构

• NodeStatusUpdater：启动时向 ResourceManager 注册，报告该节点的可用资源情况，通信的端口和后续状态的维护。
• ContainerManager：接收 RPC 请求（启动、停止），资源本地换（下载应用需要的资源到本地，根据需要共享这些资源）
  • PUBLIC:/filecache
  • PRIVATE:/usercache//filecache
  • APPLICATION:/usercache//appcache//（在程序完成后会被删除）
• ContainersLauncher：加载或终止 Container
• ContainerMonitor：监控 Container 的运行和资源使用情况
• ContainerExecutor：和底层操作系统交互，加载要运行的程序

YARN ApplicationMaster

单个作业的资源管理和任务监控。

功能描述

1. 计算应用的资源需求，资源可以是静态或动态计算的，静态的一般是 Client 申请时就指定了，动态则需要ApplicationMaster根据应用的运行状态来决定。
2. 根据数据来申请对应位置的资源 （Data Locality）
3. 向 ResourceManager 申请资源，与 NodeManager 交互进行程序的运行和监控，监控申请资源的使用情况，监控作业进度。
4. 跟踪任务状态和进度，定时向 ResourceManager 发送心跳消息，报告资源的使用情况和应用的进度信息
5. 负责本作业内任务的容错

ApplicationMaster 可以是用任何语言编写的程序，它和 ResourceManager 和 NodeManager 之间是通过 ProtocolBuf 交互，以前是一个全局 JobTracker 负责的，现在每个作业都有一个，可伸缩性更强，至少不会应为作业太多，造成 JobTracker 瓶颈。同时将作业的逻辑放到一个独立的 ApplicationMaster 中，使得灵活性更高，每个作业都可以有自己的处理方式，不用绑定到 MapReduce 的处理模式上。

如何计算资源需求

一般的 MapReduce 是根据 Block 数量来决定 Map 和 Reduce 的计算数量，然后一般的 Map 或 Reduce 就占用一个 Container。

如何发现数据的本地化

数据的本地化是通过 HDFS 的 Block 分片信息获取的。

YARN Container

1. 基本的资源单位（CPU、内存等）
2. Container 可以加载任意程序，而且不限于 Java
3. 一个 Node 可以包含多个 Container，也可以是一个大的 Container
4. ApplicationMaster 可以根据需要，动态申请和释放 Container

YARN Failover

失败类型

1. 程序问题
2. 进程奔溃
3. 硬件问题

失败处理

任务失败

1. 运行时异常或者 JVM 退出都会报告给 ApplicationMaster。
2. 通过心跳来检查挂住的任务（Timeout），会检查多次（可配置）才判断任务是否失效。
3. 一个作业的任务失败率超过配置，则认为改作业失败。
4. 失败的任务或作业都会有 ApplicationMaster 重新运行。

ApplicationMaster 失败

1. ApplicationMaster 定时发送心跳信息到 ResourceManager，通常一旦 ApplicationMaster 失败，则认为失败，但也可以通过配置多次后才失败。
2. 一旦失败，ResourceManager 会启动一个新的 ApplicationMaster。
3. 新的 ApplicationMaster 负责恢复之前错误的 ApplicationMaster 状态（yarn.app.mapreduce.am.job.recovery.enable=true），这一步是通过将应用运行状态保存到共享的存储上来实现的，ResourceManager不会负责任务状态的保存和恢复。
4. Client 也会定时向 ApplicationMaster 查询进度和状态，一旦发现其失败，则向 ResourceManager 询问新的 ApplicationMaster

NodeManager 失败

1. NodeManager 定时发送心跳到 ResourceManager，如果超过一段时间没有收到心跳信息，ResourceManager 就会将其移除。
2. 任何运行在 NodeManager 上的任务和 ApplicationMaster 都会在其他 NodeManager 上进行恢复。
3. 如果某个 NodeManager 失败的次数太多，ApplicationMaster 会将其加入黑名单（ResourceManager 没有），任务调度时不再其上运行。

ResourceManager 失败

1. 通过 checkpoint 机制，定时将其状态保存到磁盘，然后失败的时候，重新运行。
2. 通过 Zookeeper 同步状态和实现透明的高可用（HA）

可以看出，一般的错误处理都是由当前模块的父模块进行监控（心跳）和恢复。而最顶端的模块则通过定时保存，同步状态和Zookeeper来实现 HA。

Hadoop MapReduce

简介

一种分布式的计算方式，指定一个 Map（映射） 函数，用来把一组键值对映射成一组新的键值对，指定并发的 Reduce（归约） 函数，用来保证所有映射的键值对中的每一个共享相同的键组。

Map 输出格式和 Reduce 输入格式一定是相同的。

基本流程

1. 读取文件数据
2. 进行 Map 处理
3. 进行 Reduce 处理
4. 把处理结果写到文件中

详细流程

多节点下的流程

主要过程

Map Side

• Record Reader

  记录阅读器会翻译由输入格式生成的记录，记录阅读器用于数据解析给记录，并不分析记录本身。记录读取器的目的是将数据解析成记录，但不分析记录本身。它将数据以键值对的形式传输给 Mapper。通常键是位置信息，值是构成记录的数据存储块（自定义的记录不再本文的讨论范围内）。

• Map

  在映射器中用户提供的代码称为中间对。对于键值的具体定义是慎重的，因为定义对于分布式任务的完成具有重要意义。键决定可数据分类的依据，而值决定了处理器中的分析信息。

• Shuffle and Sort

  Reduce 任务以随机和排序步骤开始，此步骤写入输出文件并下载到本地计算机。这些数据采用键进行排序以把等价秘钥组合到一起。

• Reduce

  Reduce 采用分组数据作为输入，该功能传递键和此键相关值的迭代器。可以采用多种方式来汇总、过滤或者合并数据。当 Reduce 功能完成，就会发送0个或多个键值对。

• 输出格式

  输出格式会转换最终的键值对并写入文件。默认情况下键和值以 tab 分割，各记录以换行符分割。因此可以自定义更多输出格式，最终数据会写入 HDFS。类似记录读取（自定义的输出格式不再本文的讨论范围内）。

MapReduce 读取数据

通过 InputFormat 决定读取的数据的类型，然后拆分成一个个 InputSplit，每个 InputSplit 对应一个 Map 处理，RecordReader 读取 InputSplit 的内容给 Map。

InputFormat

决定读取数据的格式，可以使文件或者数据库等。

功能

1. 验证作业输入的正确性，如格式等。
2. 将输入文件切割成逻辑分片（InputSplit），一个 InputSplit 将会被分配给一个独立的 Map 任务。
3. 提供 RecorderReader 实现，读取 InputSplit 中的 K-V对 供 Mapper 使用。

方法

• List getSplits()：获取由输入文件计算出输入分片（InputSplit），解决数据或文件分割成片问题。
• RecordReader <K,V> createRecordReader()：创建 RecordReader，从InputSplit中读取数据，解决读取分片中数据问题。

类结构

• TextInputFormat：输入文件中的每一行就是一个记录，Key 是这一行的 byte offset，而 Value 是这一行的内容。
• KeyValueTextFormat：输入文件中每一行就是一个记录，第一个分隔符字符切分每行。在分隔符字符之前的内容为 Key，在之后的为 Value。分隔符变量通过 key.value.separator.in.input.line 变量设置，默认为 \t 字符。
• NLineInputFormat：与 TextInputFormat 一样，但每个数据块必须保证有且只有 N 行，mapred.line.input.format.linespermap 属性，默认为 1。
• SequenceFileInputFormat：一个用来读取字符流数据的 InputFormat，<Key,Value> 为用户自定义的。字符流数据是 Hadoop自定义的压缩的二进制数据格式。它用来优化一个从 MapReduce 任务输出到另一个 MapReduce 任务的输入之间的数据传输过程。

InputSplit

代表一个个逻辑分片，并没有真正存储数据，只是提供了一个如何将数据分片的方法。

Split内有 Location 信息，有利于数据局部化。一个InputSplit交给一个单独的 Map 处理。

public abstract class InputSplit 
    /**
     * 获取 Split 的大小，支持根据 size 对 InputSplit 排序。
     */
    public abstract long getLength() throws IOException, InterruptedException;
    
    /**
     * 获取存储该分片的数据所在的节点位置。
     */
    public abstract String[] getLocations() throws IOException, InterruptedException;

RecordReader

将 InputSplit 拆分成一个个 <Key,Value>对给 Map 处理，也是实际的文件读取分割对象。

常见问题

大文件如何处理

CombineFileInputFormat 可以将若干个 Split 打包成一个，目的是避免过多的 Map 任务（因为 Split 的数目决定了 Map 的数目，大量的 Mapper Task 创建销毁开销将是巨大的）。

如何计算 split

通常一个 split 就是一个Block（FileInputFormat 仅仅拆分比Block大的文件），这样做的好处使得 Map 可以存储有当前数据的节点上运行的本地的任务，而不需要通过网络进行跨节点的任务调度。

通过mapred.min.split.size, mapred.max.split.size，block.size 来控制拆分的大小。

如果mapred.min.split.size 大于 block size，则会将两个 Block 合成到一个 split，这样有部分 Block 数据需要通过网络读取。

如果mapred.max.split.size 小于 block size，则会将一个 Block 拆成多个 split，增加了 Map 任务数（Map 对 split 进行计算，且上报结果，关闭当前计算打开新的 split 均需要耗费资源）。

先获取文件在 HDFS 上的路径和 Block 信息，然后根据 splitSize 对文件进行切分（ splitSize = computeSplitSize(blockSize,minSize,maxSize) )，默认 splitSize 就等于 blockSize 的默认值（64M）。

public List<InputSplit> getSplits(JobContext job) throws IOException 
    // 首先计算分片的最大和最小值。这两个值将会用来计算分片的大小
    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
    long maxSize = getMaxSplitSize(job);
    
    // 拆分 splits
    List<InputSplit> splits = new ArrayList<InputSplit>();
    List<FileStatus> files = listStatus(job);
    for( FileStatus file : files )
        Path path = file.getPath();
        long length = file.getLen();
        if( length != 0 )
            FileSystem fs = path.getFileSystem(job.getConfiguration());
            // 获取该文件所有的 Block 信息列表[hostname, offset, length]
            BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
            // 判断文件是否可分割，通常是可分割的，但如果文件是压缩的，将不可分割
            if( isSplitable(job, path) )
                long blockSize = file.getBlockSize();
                // 计算分片大小，即 Math.max(minSize, Math.min(maxSize, blockSize))；
                long splitSize = computeSplitSize(blockSize, minSize, maxSize);
                
                long bytesRemaining = length;
                // 循环分片：当剩余数据与分片大小比值大于 Split_Slot 时，继续分片；
                // 小于等于时，停止分片。
                while( ((double) bytesRemaining) / splitSize > SPLIT_SLOP )
                    int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
                    splits.add(makeSplit(path, length-bytesRemaining, splitSize, blkLocations[blkIndex].getHosts()));
                    bytesRemaining -= splitSize;
                
                // 处理剩下的数据，不足一个 Block 大小的
                if(bytesRemaining != 0)
                    splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));
                
             else 
                // 不可拆分成 splits，整块返回
                splits.add(makeSplit(path, 0, length, blkLocation[0].getHosts()));
            
         else 
            // 对于长度为 0 的文件，创建空 Hosts 列表，返回
            splits.add(makeSplit(path, 0, length, new String[0]));
        
    
    // 设置输入文件的数量
    job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
    LOG.debug("Total  of Splits:" + splits.size);
    return splits;

分片间的数据如何处理

Split 是根据文件大小分割的，而一般处理是根据分隔符进行分割的，这样势必存在一条记录横跨两个 split。

解决办法：只要不是第一个 split，都会远程读取一条记录，忽略掉第一条记录。

public class LineRecordReader extends RecordReader<LongWritable, Text> 
    public static final String MAX_LINE_LENGTH = "mapreduce.input.linerecordreader.line.maxlength";
    private CompressionCodecFactory compressionCodecs = null;
   
    private long start;
    private long pos;
    private long end;
    private LineReader in;
    private int maxLineLength;
    private LongWritable key = null;
    private Text value = null;
    
    // initialize 函数即对 LineRecordReader 的一个初始化
    // 主要是计算分片的始末位置，打开输入流以供读取 K-V对，处理经过分片压缩的情况等
    public void initialize(InputSplit genericSplit, TaskAttemptContext context) throw IOException 
        FileSplit split = (FileSplit) genericSplit;
        Configuration job = context.getConfiguration();
        this.maxLineLength = job.getInt(MAX_LINE_LENGTH, Integer.Max_VALUE);
        start = split.getStart();
        end = start + split.getLength();
        final Path file = split.getPath();
        compressionCodecs = new CompressionCodecFactory(job);
        
        // 打开文件，并定位到分片读取的起始位置
        FileSystem fs = file.getFileSystem(job);
        FSDataInputStream fileIn = fs.open(split.getPath());
        
        boolean skipFirstLine = false;
        if(codec != null)
            // 文件是压缩文件的话，直接打开文件
            in = new LineReader(codec.createInputStream(fileIn),job);
            end = Lone.MAX_VALUE;
         else 
            // 只要不是第一个 split，则忽略本 split 的第一行数据
            if(start != 0)
                skipFirstLine = true;
                --start;
                // 定位到偏移位置，下面的读取就会从偏移位置开始
                fileIn.seek(start);
            
            in = new LineReader(fileIn, job);
        
        
        if(skipFirstLine) 
            // 忽略第一行数据，重新定位 start
            start += in.readLine(new Text(), 0, (int)Math.min((long) Integer.MAX_VALUE, end-start));
        
        this.pos = start;
    
    
    public boolean nextKeyValue() throws IOException 
        if(key == null) 
            key = new LongWritable();
        
        // key 即为偏移量
        key.set(pos);
        if(value == null)
            value = new Text();
        
        int newSize = 0;
        while(pos < end)
            newSize = in.readLine(value, maxLineLength, Math.max((int) Math.min(Integer.MAX_VALUE, end-pos), maxLineLength));
            // 读取数据长度为 0，则说明已经读完
            if(newSize == 0)
                break;
            
            pos += newSize;
            // 读取的数据长度小于最大行长度，也说明已经读取完毕
            if(newSize < maxLineLength)
                break;
            
            // 执行到此处，说明改行数据没读完，继续读入
        
        
        if(newSize == 0)
            key = null;
            value = null;
            return false;
         else 
            return true;
        
    

MapReduce Mapper

主要是读取InputSplit 的每一个Key-Value对，并进行处理。

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> 
    /**
     * 预处理，仅在 map task 启动时运行一次
     */
    protected void setup(Context context) throws IOException, InterruptedException
    
    
    /**
     * 对于InputSplit中的每一对<key, value>都会运行一次
     */
    @SuppressWarnings("unchecked")
    protected void map(KEYIN key, VALUEIN value, Context context) throws IOException, InterruptedException 
        context.write((KEYOUT) key, (VALUEOUT) value);
    

    /**
     * 扫尾工作，比如关闭流等
     */
    protected void cleanup(Context context) throws IOException, InterruptedException 
    
    
    /**
     * map task的驱动器
     */
    public void run(Context context) throws IOException, InterruptedException 
        setup(context);
        while (context.nextKeyValue()) 
            map(context.getCurrentKey(), context.getCurrentValue(), context);
        
        cleanup(context);
    


public class MapContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> extends TaskInputOutputContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> 
    private RecordReader<KEYIN, VALUEIN> reader;
    private InputSplit split;
    
    /**
     * Get the input split for this map.
     */
    public InputSplit getInputSplit() 
        return split;
    
    
    @Override
    public KEYIN getCurrentKey() throws IOException, InterruptedException 
        return reader.getCurrentKey();
    
    
    @Override
    public VALUEIN getCurrentValue() throws IOException, InterruptedException 
        return reader.getCurrentValue();
    

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException 
        return reader.nextKeyValue();
    

Hadoop Shuffle

对 Map 的结果进行排序并传输到 Reduce 进行处理。Map 的结果并不直接存放到硬盘，而是利用缓存做一些预排序处理。Map 会调用 Combiner 进行压缩，按 Key 进行分区、排序等，尽量减少结果的大小。每个 Map 完成后都会通知 Task，然后 Reduce 就可以进行处理。

Map 端

• 当 Map 程序开始产生结果的时候，并不是直接写到文件的，而是利用缓存做一些排序方面的预处理操作。每个 Map 任务都有一个循环内存缓冲区（默认 100M），当缓存的内容达到80%时，后台线程开始将内容写到文件，此时 Map 任务可以继续输出结果；但如果缓存区满了，Map 任务则需要等待。
• 写文件使用 round-robin 方式。在写入文件之前，现将数据按照 Reduce 进行分区。对于每一个分区，都会在内存中根据 Key 进行排序，如果配置了 Combiner，则排序后执行 Combiner（Combiner之后可以减少写入文件和传输的数据）。
• 每次结果到达缓存区的阈值时，都会创建一个文件，当 Map 结束后，可能会产生大量的文件。在 Map 完成前，会将这些文件进行合并和排序。如果文件的数量超过 3 个，则合并后会再次运行 Combiner（1/2 个文件就没有必要了）。
• 如果配置压缩，则最终写入的文件会先进行压缩，这样可以减少写入和传输的数据。一旦 Map 完成，则通知任务管理器，此时 Reduce 就可以开始复制结果数据。

Reduce 端

• Map 的结果文件存放到运行 Map 任务的机器的本地硬盘中。如果 Map 的结果很少，则直接放到内存中，否则写入文件中。
• 同时后台线程将这些文件进行合并和排序到一个更大的文件中（如果文件是压缩的，则需要先解压文件）
• 当所有的 Map 结果都被复制和合并后，就会调用 Reduce 方法，Reduce 结果会写入到 HDFS 中

调优

• 一般的原则是给 Shuffle 分配尽可能多的内存，但前提要保证 Map、Reduce 任务有足够的内存。
• 对于 Map，主要就是避免把文件写入磁盘，例如使用 Combiner，增大io.sort.mb 的值
• 对于 Reduce，主要是把 Map 的结果尽可能地保存到内存中，同样也要避免中间结果写入磁盘。默认情况下，所有的内存都是分配给 Reduce 方法的，如果 Reduce 方法不怎么消耗内存，可以将 mapred.inmem.merge.threshold 设置为 0，mapred.job.reduce.input.buffer.percent 设成 1.0。
• 在任务监控中可以通过 Spilled Records Counter 来监控写入磁盘的数，但这个值是包括 Map 和 Reduce 的。对于 I/O 方面，可以 Map 的结果可以使用压缩，同时增大 Buffer Size（io.file.buffer.size，默认4kb）。

配置

属性	默认值	描述
io.sort.mb	100	映射输出分类时所使用缓冲区的大小.
io.sort.record.percent	0.05	剩余空间用于映射输出自身记录.在1.X发布后去除此属性.随机代码用于使用映射所有内存并记录信息.
io.sort.spill.percent	0.80	针对映射输出内存缓冲和记录索引的阈值使用比例.
io.sort.factor	10	文件分类时合并流的最大数量。此属性也用于reduce。通常把数字设为100.
min.num.spills.for.combine	3	组合运行所需最小溢出文件数目.
mapred.compress.map.output	false	压缩映射输出.
mapred.map.output.compression.codec	DefaultCodec	映射输出所需的压缩解编码器.
mapred.reduce.parallel.copies	5	用于向reducer传送映射输出的线程数目.
mapred.reduce.copy.backoff	300	时间的最大数量，以秒为单位，这段时间内若reducer失败则会反复尝试传输
io.sort.factor	10	组合运行所需最大溢出文件数目.
mapred.job.shuffle.input.buffer.percent	0.70	随机复制阶段映射输出缓冲器的堆栈大小比例
mapred.job.shuffle.merge.percent	0.66	用于启动合并输出进程和磁盘传输的映射输出缓冲器的阀值使用比例
mapred.inmem.merge.threshold	1000	用于启动合并输出和磁盘传输进程的映射输出的阀值数目。小于等于0意味着没有门槛，而溢出行为由 mapred.job.shuffle.merge.percent单独管理.
mapred.job.reduce.input.buffer.percent	0.0	用于减少内存映射输出的堆栈大小比例，内存中映射大小不得超出此值。若reducer需要较少内存则可以提高该值.

Hadoop 编程

处理

1. select：直接分析输入数据，取出需要的字段数据即可
2. where：也是对输入数据处理的过程进行处理，判断是否需要该数据
3. aggregation：min，max，sum
4. group by：通过 Reduce 实现
5. sort
6. join：Map join，Reduce join

Third-Party Libraries

第一种：指定依赖可以利用Public Cache

export LIBJARS=$MYLIB/commons-lang-2.3.jar, hadoop jar prohadoop-0.0.1-SNAPSHOT.jar org.aspress.prohadoop.c3. WordCountUsingToolRunner -libjars $LIBJARS

第二种：包含依赖，则每次都需要拷贝

hadoop jar prohadoop-0.0.1-SNAPSHOT-jar-with-dependencies.jar org.aspress.prohadoop.c3. WordCountUsingToolRunner The dependent libraries are now included inside the application JAR file

Hadoop IO

1. 输入文件从 HDFS 进行读取。
2. 输出文件会存入本地磁盘。
3. Reducer 和 Mapper 间的网络I/O，从 Mapper 节点得到 Reducer 的检索文件。
4. 使用 Reducer 实例从本地磁盘会读数据。
5. Reducer 输出-回传到 HDFS。

串行化

• 传输、存储都需要
• Writable 接口
• Avro 框架：IDL，版本支持，跨语言，JSON-linke

压缩

• 能够减少磁盘的占用空间和网络传输的量
• Compressed Size，Speed，Splittable
• gzip，bzip，LZO，LZ4，Snappy
• 要比较各种压缩算法的压缩比和性能

重点：压缩和拆分一般是冲突的（压缩后的文件的 Block 是不能很好地拆分独立运行，很多时候某个文件的拆分点是被拆分成两个压缩文件中，这是 Map 任务就无法处理，所以对于这些压缩，Hadoop 往往是直接使用一个 Map 任务处理整个文件的分析。Map 的输出结果也可以进行压缩，这样可以减少 Map 结果到 Reduce 的传输的数据量，加快传输速率

完整性

• 磁盘和网络很容易出错，保证数据传输的完整性一般是通过CRC32这种校验法
• 每次写数据到磁盘前都验证一下，同时保存校验码
• 每次读取数据时，也验证校验码
• 同时每个 DataNode 都会定时检查每一个 Block 的完整性
• 当发现某个 Block 数据有问题时，也不是立刻报错，而是先去 NameNode 找一块该数据的完整备份进行恢复，不能恢复才报错

Hadoop 安装

安装方式

• 单节点安装

  所有服务都运行在一个 JVM 中，适合调试、单元测试

• 伪集群

  所有服务运行在一台机器中，每个服务都在独立的 JVM 中，适合做简单、抽样测试

• 多节点集群

  服务运行在不同的机器中，适合生产环境

配置公共账号

方便主从服务器进行无密钥通信，主要使用公钥/私钥机制。

• 所有节点的账号都一样，在主节点上执行 ssh-keygen -t rsa 生成密钥对。
• 复制公钥到每台目标节点中。

Hadoop 配置

配置文件

• xxx-default.xml：只读，默认的配置
• xxx-site.xml：替换 default 中的配置
  • core-site.xml：配置公共属性
  • hdfs-site.xml：配置HDFS
  • yarn-site.xml：配置YARN
  • mapred-site.xml：配置MapReduce

配置文件的顺序

1. 在 JobConf 中指定的
2. 客户端机器上的 xxx-site.xml中的配置
3. Slave 节点上的 xxx-site.xml 中的配置
4. xxx-default.xml 中的配置

如果某个属性不想被覆盖，可以将其设置为 final

<property>
    <name>PROPERTY_NAME</name>
    <value>PROPERTY_VALUE</value>
    <final>true</final>
</property>