mapreduce 是什么？

MapReduce是Hadoop的核心之一，可以分成Map和Reduce两部分理解。

1.Map：映射过程，把一组数据按照某种Map函数映射成新的数据。我们将这句话拆分提炼出重要信息，也就是说，map主要是：映射、变换、过滤的过程。一条数据进入map会被处理成多条数据，也就是1进N出。

2.Reduce：归纳过程，把若干组映射结果进行汇总并输出。我们同样将重要信息提炼，得到reduce主要是：分解、缩小、归纳的过程。一组数据进入reduce会被归纳为一组数据（或者多组数据），也就是一组进N出。

graph LR
    A[输入数据集] --map--> B[中间结果数据集]
    B--reduce-->C[最终结果数据集]

mapreduce的基本工作流程

分片、格式化数据源

输入 Map 阶段的数据源，必须经过分片和格式化操作。

分片操作：

指的是将源文件划分为大小相等的小数据块( Hadoop 2.x 中默认 128MB )，也就是分片( split )，
Hadoop 会为每一个分片构建一个 Map 任务，并由该任务运行自定义的 map() 函数，从而处理分片里的每一条记录;

格式化操作：

将划分好的分片( split )格式化为键值对形式的数据，其中， key 代表偏移量， value 代表每一行内容。

执行 MapTask

每个 Map 任务都有一个内存缓冲区(缓冲区大小 100MB )，输入的分片( split )数据经过 Map 任务处理后的中间结果会写入内存缓冲区中。
如果写人的数据达到内存缓冲的阈值( 80MB )，会启动一个线程将内存中的溢出数据写入磁盘，同时不影响 Map 中间结果继续写入缓冲区。
在溢写过程中， MapReduce 框架会对 key进行排序，如果中间结果比较大，会形成多个溢写文件，最后的缓冲区数据也会全部溢写入磁盘形成一个溢写文件，如果是多个溢写文件，则最后合并所有的溢写文件为一个文件。

执行 Shuffle 过程

MapReduce 工作过程中， Map 阶段处理的数据如何传递给 Reduce 阶段，这是 MapReduce 框架中关键的一个过程，这个过程叫作 Shuffle 。
Shuffle 会将 MapTask 输出的处理结果数据分发给 ReduceTask ，并在分发的过程中，对数据按 key 进行分区和排序。

执行 ReduceTask

输入 ReduceTask 的数据流是形式，用户可以自定义 reduce()方法进行逻辑处理，最终以的形式输出。

写入文件

MapReduce 框架会自动把 ReduceTask 生成的传入 OutputFormat 的 write 方法，实现文件的写入操作。

graph LR
  subgraph 整体流程
    direction TB
    A1[分片 格式化] --> A2[执行Map Task]
    A2-->A3[执行Shuffle]
    A3-->A4[执行Reduce Task]
    A4-->A5[落盘 写入文件]
  end
  整体流程 --> B[分片 格式化]
  整体流程 --> C[执行Map Task]
  整体流程-->D[执行Shuffle]
  整体流程-->E[执行Reduce Task]
  整体流程-->F[落盘 写入文件]
  B-->B1(分片 128M =>MapTask.map)
  B-->B2(格式化 split=>key,value) -->B3(exg: wordcount key 偏移量  value 每一行的内容)
  C-->C1(1 每个 Map 任务都有一个内存缓冲区 100MB )
  C-->C2(2 map处理后的结果写入内存缓存区)
  C-->C3(3 内存缓存区达到80% key排序溢写到文件)
  C-->C4(4 多个溢写文件合并)
  D-->D1(maptask=>reducetask 分发和排序)
  E-->E1(key, value-list => key value)
  F-->F1(key,value => OutputFormat.write)

MapTask 详解

Read 阶段： MapTask 通过用户编写的 RecordReader ，从输入的 InputSplit 中解析出一个个 key / value 。
Map 阶段：将解析出的 key / value 交给用户编写的 Map ()函数处理，并产生一系列新的 key / value 。
Collect 阶段：在用户编写的 map() 函数中，数据处理完成后，一般会调用 outputCollector.collect() 输出结果，在该函数内部，它会将生成的 key / value 分片(通过调用 partitioner )，并写入一个环形内存缓冲区中(该缓冲区默认大小是 100MB )。

Spill 阶段：即“溢写”，当缓冲区快要溢出时(默认达到缓冲区大小的 80 %)，会在本地文件系统创建一个溢出文件，将该缓冲区的数据写入这个文件。

将数据写入本地磁盘前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作。
写入磁盘之前，线程会根据 ReduceTask 的数量，将数据分区，一个 Reduce 任务对应一个分区的数据。
这样做的目的是为了避免有些 Reduce 任务分配到大量数据，而有些 Reduce 任务分到很少的数据，甚至没有分到数据的尴尬局面。
如果此时设置了 Combiner ，将排序后的结果进行 Combine 操作，这样做的目的是尽可能少地执行数据写入磁盘的操作。

Combine 阶段：当所有数据处理完成以后， MapTask 会对所有临时文件进行一次合并，以确保最终只会生成一个数据文件

合并的过程中会不断地进行排序和 Combine 操作，
其目的有两个：一是尽量减少每次写人磁盘的数据量;二是尽量减少下一复制阶段网络传输的数据量。
最后合并成了一个已分区且已排序的文件。

ReduceTask 详解

Copy 阶段： Reduce 会从各个 MapTask 上远程复制一片数据（每个 MapTask 传来的数据都是有序的），并针对某一片数据，如果其大小超过一定國值，则写到磁盘上，否则直接放到内存中
Merge 阶段：在远程复制数据的同时， ReduceTask 会启动两个后台线程，分别对内存和磁盘上的文件进行合并，以防止内存使用过多或者磁盘文件过多。

Sort 阶段：用户编写 reduce() 方法输入数据是按 key 进行聚集的一组数据。

为了将 key 相同的数据聚在一起， Hadoop 采用了基于排序的策略。
由于各个 MapTask 已经实现对自己的处理结果进行了局部排序，因此， ReduceTask 只需对所有数据进行一次归并排序即可。

Reduce 阶段：对排序后的键值对调用 reduce() 方法，键相等的键值对调用一次 reduce()方法，每次调用会产生零个或者多个键值对，最后把这些输出的键值对写入到 HDFS 中

Write 阶段： reduce() 函数将计算结果写到 HDFS 上。

合并的过程中会产生许多的中间文件(写入磁盘了)，但 MapReduce 会让写入磁盘的数据尽可能地少，并且最后一次合并的结果并没有写入磁盘，而是直接输入到 Reduce 函数。

https://blog.csdn.net/weixin_43542605/article/details/122288056

Hadoop HA（High Available）经过同时配置两个处于Active/Passive模式的Namenode，分别叫Active Namenode和Standby Namenode。 Standby Namenode做为热备份，从而容许在机器发生故障时可以快速进行故障转移，同时在平常维护的时候使用优雅的方式进行Namenode切换。Namenode只能配置一主一备，不能多于两个Namenode。

Hadoop HA基本流程

集群总体上可以分为三部分：NameNode集群、JournalNode集群和Zookeeper集群。NameNode在某一时刻只有一个处于活跃状态，其他的都处于standby状态；JournalNode负责把edits文件传到standby的NameNode上；Zookeeper负责监控NameNode宕机情况，ZKFC(ZookeeperFailoverController)是专门监控NameNode健康的。

为了同步NameNode的元数据一致，有专门的JournalNode来同步元数据文件，活跃的NameNode的edits文件会写入journalnode集群，其他standby的结点会去读取journalnode上的edits文件，以此来同步自身的元数据。

ZKFC的HealthyMonitor是监控NameNode的进程，是专门监控NameNode将康情况的进程。
HealthyMonitor会定时想ZKFC进程报告NameNode情况。
当HealthyMonitor出现汇报了NameNode，ZKFC就会向AcitveStandbyEloctor报告。
AcitveStandbyEloctor接到NameNode宕机报告就会通知zk集群选举出新的NameNode。
zk集群经过内部选举，返回一个standby的NameNode给AcitveStandbyEloctor。
AcitveStandbyEloctor想ZKFC报告选举结果。
ZKFC为了防止是网络原因导致NameNode假死，就会结束NameNode进程。
zk集群就会通知另一个ZKFC要求它修改它监控的NameNode的进程为活跃节点。

HA技术关键点

HA问题中需要解决的两个问题：

元数据一致性：Standby节点和Active节点的元数据一致性。
主备自动切换：Active节点服务中断时，Standby节点可以立即启动对外提供服务。

为了确保故障转移可以快速完成，Standby Namenode须要维护最新的Block位置信息，即每一个Block副本存放在集群中的哪些节点上。为了达到这一点，Datanode同时配置主备两个Namenode，并同时发送Block报告和心跳到两台Namenode。

确保任什么时候刻只有一个Namenode处于Active状态很是重要，不然可能出现数据丢失或者数据损坏。当两台Namenode都认为本身的Active Namenode时，会同时尝试写入数据（不会再去检测和同步数据）。为了防止这种脑裂现象，Journal Nodes只容许一个Namenode写入数据，内部经过维护epoch数来控制，从而安全地进行故障转移。

HA其本质上就是要保证主备NN元数据是保持一致的，即保证fsimage和editlog在备NN上也是完整的。元数据的同步很大程度取决于EditLog的同步，而这步骤的关键就是共享文件系统

有两种方式能够进行edit log共享：

使用QJM（Quorum Journal Manager）共享edit log
使用NFS（Network File System）共享edit log（存储在NAS/SAN）

NFS的方式

all name space edits logged to shared storage;Block reports are sent to both name nodes

显然NFS做为主备Namenode的共享存储。这种方案可能会出现脑裂（split-brain），即两个节点都认为本身是主Namenode并尝试向edit log写入数据，这可能会致使数据损坏。经过配置fencin脚原本解决这个问题，fencing脚本用于：

将以前的Namenode关机
禁止以前的Namenode继续访问共享的edit log文件

使用这种方案，管理员就能够手工触发Namenode切换，而后进行升级维护。但这种方式存在如下问题：

只能手动进行故障转移，每次故障都要求管理员采起措施切换。
NAS/SAN设置部署复杂，容易出错，且NAS自己是单点故障。
Fencing 很复杂，常常会配置错误。
没法解决意外（unplanned）事故，如硬件或者软件故障。oop
所以须要另外一种方式来处理这些问题：
自动故障转移（引入ZooKeeper达到自动化）
移除对外界软件硬件的依赖（NAS/SAN）
同时解决意外事故及平常维护致使的不可用

Quorum Journal Manager

QJM（Quorum Journal Manager）是Hadoop专门为Namenode共享存储开发的组件，一般是奇数点结点组成。其集群运行一组Journal Node，每一个Journal 节点暴露一个简单的RPC接口，容许Namenode读取和写入数据，数据存放在Journal节点的本地磁盘。当Namenode写入edit log时，NameNode会同时向所有JournalNode并行写文件，当超过半数节点回复确认成功写入以后，edit log就认为是成功写入。

1、 QJM写过程

NameNode 会把 EditLog 同时写到本地和 JournalNode 中。写本地由配置中的参数dfs.namenode.name.dir来控制，写JN由参数dfs.namenode.shared.edits.dir控制，在写EditLog时会由两个不同的输出流来控制日志的写过程，分别是：

EditLogFileOutputStream（本地输出流）
QuorumOutputStream（JN输出流）

NameNode在写EditLog时，并不是直接写到磁盘中，为保证高吞吐，NameNode会分别为EditLogFileOutputStream和QuorumOutputStream定义两个同等大小的Buffer，大小大概是512KB，一个写Buffer（buffCurrent），一个同步Buffer（buffReady），这样可以一边写一边同步，所以EditLog是一个异步写过程，同时也是一个批量同步的过程，避免每写一笔就同步一次日志。

这个是怎么实现边写边同步的呢，这中间其实是有一个缓冲区交换的过程，即bufferCurrent和buffReady在达到条件时会触发交换，如bufferCurrent在达到阈值同时bufferReady的数据又同步完时，bufferReady数据会清空，同时会将bufferCurrent指针指向bufferReady以满足继续写，另外会将bufferReady指针指向bufferCurrent以提供继续同步EditLog。

flowchart TD
    A((Client)) --修改--> B(NameNode)
    B-->C(本地 \n EditLogFileOutputStream)
    B-->D(JournalNode \n QuorumOutputStream)
    C-->E(bufferCurrent)
    C-->F(bufferReady)
    D-->H(bufferCurrent)
    D-->I(bufferReady)
    E-->M(本地目录)
    F-->M
    H-->N(JournalNode)
    I-->N

这里有一个问题，既然EditLog是异步写的，怎么保证缓存中的数据不丢呢，其实这里虽然是异步,但实际所有日志都需要通过logSync同步成功后才会给client返回成功码，假设某一时刻NameNode不可用了，其内存中的数据其实是未同步成功的，所以client会认为这部分数据未写成功。

（1）隔离双写

在Active NN每次同步EditLog到JN时，先要保证不会有两个NN同时向JN同步日志。这涉及一个很重要的概念Epoch Numbers，很多分布式系统都会用到。

成为Active结点时，其会被赋予一个EpochNumber，每个EpochNumber是惟一的，不会有相同的EpochNumber出现。EpochNumber有严格顺序保证，每次NN切换后其EpochNumber都会自增1，后面生成的EpochNumber都会大于前面的EpochNumber。QJM是怎么保证上面特性的呢，主要有以下几点：

在对EditLog作任何修改前，QJM(NameNode上)必须被赋予一个EpochNumber；
QJM把自己的EpochNumber通过newEpoch(N)的方式发送给所有JN结点；
当JN收到newEpoch请求后，会把QJM的EpochNumber保存到一个lastPromisedEpoch变量中并持久化到本地磁盘；
ANN同步日志到JN的任何RPC请求（如logEdits(),startLogSegment()等），都必须包含ANN的EpochNumber；
JN在收到RPC请求后，会将之与lastPromisedEpoch对比，如果请求的EpochNumber小于lastPromisedEpoch,将会拒绝同步请求，反之，会接受同步请求并将请求的EpochNumber保存在lastPromisedEpoch；

这样就能保证主备NN发生切换时，就算同时向JN同步日志，也能保证日志不会写乱，因为发生切换后，原ANN的EpochNumber肯定是小于新ANN的EpochNumber，所以原ANN向JN的发起的所有同步请求都会拒绝，实现隔离功能，防止了脑裂。

（2）恢复in-process日志

如果在写过程中写失败了，可能各个JN上的EditLog的长度都不一样，需要在开始写之前将不一致的部分恢复。恢复机制如下：

Active NN先向所有JN发送getJournalState请求；
JN会向ANN返回一个Epoch（lastPromisedEpoch)；
Active NN收到大多数JN的Epoch后，选择最大的一个并加1作为当前新的Epoch，然后向JN发送新的newEpoch请求，把新的Epoch下发给JN；
JN收到新的Epoch后，和lastPromisedEpoch对比，若更大则更新到本地并返回给Active NN自己本地一个最新EditLogSegment起始事务Id,若小则返回NN错误；
Active NN收到多数JN成功响应后认为Epoch生成成功，开始准备日志恢复；
Active NN会选择一个最大的EditLogSegment事务ID作为恢复依据，然后向JN发送prepareRecovery； RPC请求，对应Paxos协议2p阶段的Phase1a，若多数JN响应prepareRecovery成功，则可认为Phase1a阶段成功；
Active NN选择进行同步的数据源，向JN发送acceptRecovery RPC请求，并将数据源作为参数传给JN。
JN收到acceptRecovery请求后，会从JournalNodeHttpServer下载EditLogSegment并替换到本地保存的EditLogSegment，对应Paxos协议2p阶段的Phase1b，完成后返回Active NN请求成功状态。
Active NN收到多数JN的响应成功请求后，向JN发送finalizeLogSegment请求，表示数据恢复完成，这样之后所有JN上的日志就能保持一致。数据恢复后，Active NN上会将本地处于in-process状态的日志更名为finalized状态的日志，形式如editsstart-txidstop-txid。

（3）日志同步

执行logSync过程，将ANN上的日志数据放到缓存队列中
将缓存中数据同步到JN，JN有相应线程来处理logEdits请求
JN收到数据后，先确认EpochNumber是否合法，再验证日志事务ID是否正常，将日志刷到磁盘，返回ANN成功码
ANN收到JN成功请求后返回client写成功标识，若失败则抛出异常

通过上面一些步骤，日志能保证成功同步到JN，同时保证JN日志的一致性，进而备NN上同步日志时也能保证数据是完整和一致的。

2、QJM读过程

读过程是面向备NN(Standby NN)的，Standby NN定期检查JournalNode上EditLog的变化，然后将EditLog拉回本地。Standby NN上有一个线程StandbyCheckpointer，会定期将Standby NN上FSImage和EditLog合并，并将合并完的FSImage文件传回主NN（Active NN）上，就是所说的Checkpointing过程。下面我们来看下Checkpointing是怎么进行的。

在2.x版本中，已经将原来的由SecondaryNameNode主导的Checkpointing替换成由Standby NN主导的Checkpointing。

在Standby NN上先检查前置条件，前置条件包括两个方面：距离上次Checkpointing的时间间隔和EditLog中事务条数限制。
前置条件任何一个满足都会触发Checkpointing，然后SNN会将最新的NameSpace数据即SNN内存中当前状态的元数据保存到一个临时的fsimage文件( fsimage.ckpt）
然后比对从JN上拉到的最新EditLog的事务ID，将fsimage.ckpt_中没有，EditLog中有的所有元数据修改记录合并一起并重命名成新的fsimage文件，同时生成一个md5文件。
将最新的fsimage再通过HTTP请求传回ANN。

通过定期合并fsimage有什么好处？

可以避免EditLog越来越大，合并成新fsimage后可以将老的EditLog删除
可以避免主NN（ANN）压力过大，合并是在SNN上进行的
可以保证fsimage保存的是一份最新的元数据，故障恢复时避免数据丢失

三、主备自动切换
Hadoop的主备选举依赖于ZooKeeper。
整个切换过程是由ZKFC来控制的，ZKFC是实现主备切换的组件。每个运行的NameNode上都会有一个ZKFC进程（实际是一个Hadoop进程）。主要的功能如下：

健康检测：ZKFC会使用健康检测命令定期的ping同节点中的NameNode，只要该NameNode及时的回复健康，则任务当前NameNode是健康的；
Zookeeper会话管理: 当本地NameNode是健康的，ZKFC会保持一个在Zookeeper中打开的会话。如果本地NameNode处于Active状态，ZKFC会保持一个特殊的znode锁，如果回话中断，锁节点讲自动删除；
基于Zookeeper的选举: 如果本地的NameNode是健康的，且ZKFC发现没有其他的节点持有当前的znode锁，它会为自己获取该锁。如果成功则进行故障切换，并且确保之前的NameNode的进程中断，将本地NameNode切换为Active；

在故障切换期间，ZooKeeper主要是发挥什么作用有以下几点：

失败保护：集群中每一个NameNode都会在ZooKeeper维护一个持久的session，机器一旦挂掉，session就会过期，故障迁移就会触发；
Active NameNode选择：ZooKeeper有一个选择ActiveNN的机制，一旦现有的ANN宕机，其他NameNode可以向ZooKeeper申请排他成为下一个Active节点；
防脑裂： ZK本身是强一致和高可用的，可以用它来保证同一时刻只有一个活动节点；

参考链接：
https://blog.csdn.net/weixin_43854618/article/details/108808274

https://blog.csdn.net/shan19920501/article/details/124911283

flink 一步一步学习

flink是一个用于对无界和有界数据流进行有状态计算的框架和分布式处理引擎，被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算，即它是一个持高吞吐、低延迟、高性能的分布式处理框架。是Apache软件基金会的顶级项目

flink是什么？

对于技术学习，无论是什么语言什么工具，对其介绍或者说学习最主要的途径就是官方网站和主要文档。其它任何地方的信息都是其官方网站的信息挪用。对flink的介绍在其官网上有下边一段话。

Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams. Flink has been designed to run in all common cluster environments, perform computations at in-memory speed and at any scale.

大概翻译一下： flink是一个用于对无界和有界数据流进行有状态计算的框架和分布式处理引擎，被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算，即它是一个持高吞吐、低延迟、高性能的分布式处理框架

无界数据流：无界数据流有一个开始但是没有结束
有界数据流：有界数据流有明确定义的开始和结束

flink的特点

用一句话来概述特点就是，flink基于数据流的状态计算。而在其官网地址打开时有一张图也可以具体的体现其特点。事件驱动、流处理、流批一体
flink的基本特点介绍

flink VS spark

在spark中，一切都是由批次组成的，离线数据是一个大批次，而实时数据是由一个一个无限的小批次组成的，本质是RDD,是数据集合，是有界的。
在flink中，一切都是由流组成的，离线数据是有界限的流，实时数据是一个没有界限的流，这就是所谓的有界流和无界流。

标签： hadoop

MapReduce 基础工作原理