Hadoop

Hadoop

Hadoop1.x 2.x 3.x区别

Hadoop1.x组成：MapReduce负责计算和资源调度，HDFS负责数据存储，Common辅助工具。

Hadoop2.x组成：MapReduce负责计算，Yarn负责资源调度，HDFS负责数据存储，Common是辅助工具。

Hadoop3.x在组成上2.x没有变化，在功能上进行了改进：[Hadoop2与Hadoop3的区别 - 洛柯 - 博客园 (cnblogs.com)](https://www.cnblogs.com/-luoke/p/12809623.html#:~:text=Hadoop 2.x - 可以通过复制（浪费空间）来处理容错。,Hadoop 3.x - 可以通过Erasure编码处理容错。 4.数据平衡)

Hadoop是什么

1.是一个分布式系统基础架构

2.用于解决海量数据的存储和分析计算问题

Hadoop特点

高可靠性：维护多个副本，即使某个节点出现故障，也不会导致数据丢失

高扩展性：在集群间分配任务数据，方便扩展结点

高效性：MapReduce的思想，加快任务处理速度

高容错性：自动重新分配失败的任务

Hadoop序列化

序列化：把内存中的对象，转换成字节序列（或者其他数据传输协议）以便存储到磁盘和网络传输。

反序列化：将收到的字节序列（或者其他数据传输协议）或磁盘的持久化数据，转换成内存中的对象。

为什么要序列化：对象被序列化后可以用于存储和发送到其他计算机。

java的序列化:java的序列化是重量级序列化框架，对象被序列化后附带许多额外信息，不适合在网络中高效传输。

Hadoop自己开发了一套序列化机制：Writable

序列化的步骤：

（1）必须实现Writable接口

（2）反序列化时，需要反射调用空参构造函数，所以必须有空参构造

（3）重写序列化方法

（4）重写反序列化方法

（5）注意反序列化的顺序和序列化的顺序完全一致

（6）要想把结果显示在文件中，需要重写toString()，可用”\t”分开，方便后续用。

（7）如果需要将自定义的bean放在key中传输，则还需要实现Comparable接口，因为MapReduce框中的Shuffle过程要求对key必须能排序。

为什么要有无参构造：反序列化时需要使用反射，使用反射时，Class.newInstance()方法要求该Class对应类有无参构造方法。但如果直接不管，也会默认有无参，只要不用声明，就不用管（2）了。

HDFS

HDFS特点

分布式文件管理系统，适用于一次写入，多次读出的场景，不支持文件修改，但支持追加

优点	缺点
高容错性	不适合低时延数据访问
适合处理大量数据	无法高效对大量小文件进行存储
造价低	不支持并发写入和文件随机修改

为什么无法高效对大量小文件进行存储：

​ 每个小文件都需要占用NameNode来存储文件目录和块信息，每个文件存放信息大约会占用150kb的内存。一方面是占用内存过多，另一方面是小文件过多时，小文件的寻址时间会超过读取时间。

为什么可以追加但不支持修改：因为修改的效率极慢，每次修改都会进行一次I/O操作，违背了Hadoop高效性的原则。因此HDFS适合用于做数据分析，但不适合做网盘。

HDFS组成

NameNode：用于管理HDFS的名称空间、配置副本策略、管理数据块映射信息、处理客户端读写请求。

DataNode：存储实际的数据块，执行数据块的读/写

Client：文件切分（上传文件时，Client将文件切分为一个个的Block，然后上传）；与NameNode交互，获取文件的位置信息；与DataNode交互，读/写数据；提供一些管理HDFS的命令；提供命令访问HDFS(如对HDFS的CRUD)

SecondaryNode：辅助NameNode，分担其工作量（如定期合并Fsimage和Edits）；紧急情况下可辅助恢复NameNode

HDFS数据块

HDFS物理上分块存储，块大小可通过参数配置规定，2.x和3.x中默认大小128MB，1.x是64MB。数据块是HDFS的存储单位

为什么是128MB：（1）块不能太大（块太大了会导致传输时间明显大于寻址时间，且Map处理的运行速度会变慢）；（2）块不能太小（块太小时寻址时间会明显增大，寻找到block开始位置的时间过长）；（3）磁盘传输速率影响(HDFS的平均寻址时间为10ms，且寻址时间为传输时间的1%时最佳，则：10ms/1%=1s，最佳传输时间为1s。由于目前磁盘传输速率普遍为100MB/s：100MB*1S=100MB,所以设置为128MB最佳)

但由于实际开发中公司的传输速率更大，因此可以将块大小适量设置为256MB等。

HDFS写

（1）客户端通过Distributed FileSystem模块向NameNode请求上传文件，NameNode检查目标文件是否已存在，父目录是否存在。

（2）NameNode返回是否可以上传。

（3）客户端请求第一个 Block上传到哪几个DataNode服务器上。

（4）NameNode返回3个DataNode节点：dn1、dn2、dn3。

（5）客户端通过FSDataOutputStream模块请求dn1上传数据，dn1收到请求继续调用dn2，dn2再调用dn3，将这个通信管道建立完成。

（6）dn1、dn2、dn3逐级应答客户端。

（7）客户端开始往dn1上传第一个Block（先从磁盘读取数据放到一个本地内存缓存），以Packet为单位，dn1收到一个Packet就会传给dn2，dn2传给dn3；dn1每传一个packet会放入一个应答队列等待应答。

（8）当一个Block传输完成之后，客户端再次请求NameNode上传第二个Block的服务器。（重复执行3-7步）。

节点距离怎么计算：HDFS写过程会选择待上传数据最近距离的DataNode接收，会使用机架感知。同节点距离为0，同机架的距离为2，同集群为4，不同数据中心为6。

节点选择：第一个副本在Client所在节点上，如果客户端在集群外，随机选一个。第二个副本在另一个机架的随机节点，第三个副本在第二个副本所在机架的另一个随机节点。

为什么第三个副本和第二个副本在同一个机架？：提高数据的可靠性，同时在一定程度上保证数据的传输速率。

为什么Client以串行的方式建立通道？：降低I/O开销

数据传输的时候如何保证数据成功？：Block并非最小传输单元，在传输过程中会以Packet的形式进行数据传输，每个Packet为64MB,packet将数据传入应答队列，应答队列依次将数据传入ack队列，采用ack回执的策略来保证数据能完整上传成功。

NN和2NN

NameNode中的元数据存储位置：

元数据既需要放在内存中，也需要放在磁盘中(fsImage)。为了保证内存与FsImage的同步效率，引入了Edits文件，元数据中有更新时就追加进Edits。需要定期对FsImage和Edits进行合并，因此引入了SecondaryNameNode。每隔一个小时或每当NN的Edits操作次数达到100万次(每隔60秒2NN会检测一次NN方的edits文件的操作次数)，即对两者进行合并。

DataNode工作机制

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-oKPcUsos-1676377114692)(F:./DataNode工作机制.png)]

数据在DataNode上以文件形式存储在磁盘，包含两个文件：数据本身和元数据（数据块长度、块数据校验和、时间戳）

数据如何保证完整性：DataNode读取Block的时候会计算校验和（CheckSum），如果计算后的CheckSum，与Block创建时值不一样，说明Block已经损坏（校验算法：crc（32），md5（128），sha1（160））。DataNode在其文件创建后周期验证CheckSum。

DataNode掉线参数设置：DataNode进程死亡或网络故障造成DataNode和NameNode无法通信，NameNode等待一段时间后将节点认定为死亡。

TimeOut=2*dfs.namenode.heartbeat.recheck-interval+10*dfs.heartbeat.interval

dfs.namenode.heartbeat.recheck-interval默认为5分钟，dfs.heartbeat.interval默认为3秒，默认TimeOut=10分30秒

节点的服役与退役

服役只需要将新的节点启动起来就可以了

退役需要在NameNode所在节点的/opt/module/hadoop-3.1.3/etc/hadoop目录下分别创建whitelist 和blacklist文件，然后在hdfs-site.xml配置文件中增加dfs.hosts和dfs.hosts.exclude配置参数，重启集群。

如果副本数是3，且服役的节点小于等于3，则不能退役成功，需要先修改副本数才能退役。且一个节点不能同时出现在白名单和黑名单中。

如果数据不平衡了，可以使用sbin/start-balancer.sh实现集群的平衡

MapReduce

MapReduce是分布式运算程序的编程框架。将计算分为Map和Reduce两个阶段。核心功能是将用户编写的逻辑代码和自带默认组件整合成一个完整的分布式运算程序。Map阶段并行处理输入数据，Reduce阶段对Map结果进行汇总。

优点	缺点
易于编程	不删除实时计算
良好的扩展性	不擅长流式计算
高容错性	不擅长DAG（有向无环图）计算
适合海量数据离线处理

为什么不适合流式计算：流式计算的输入数据是动态的，而MapReduce输入数据是静态的。

为什么不适合DAG计算：MR不适合用于多个应用程序存在依赖关系的场景。每个MR的输出结果都写入磁盘，会造成大量的磁盘IO，导致性能低下

MR编程模型只能包含一个Map阶段和一个Reduce阶段，如果用户逻辑复杂，则需要多个MR串行

一个完整的MapReduce程序在分布式运行时有三类实例进程：

（1）MrAppMaster：负责整个程序的过程调度及状态协调。

（2）MapTask：负责Map阶段的整个数据处理流程。

（3）ReduceTask：负责Reduce阶段的整个数据处理流程。

MR编程规范

Mapper 、Reducer、Driver

Mapper：

​ （1）用户自定义的Mapper继承自己的父类

​ （2）Mapper的输入数据是KV的形式（KV类型可自定义）

​ （3）Mapper中的业务逻辑写在map()方法中

​ （4）Mapper的输出数据是KV的形式

​ （5）map()方法对每个<K,V>调用一次

Reducer:

​ (1) 用户自定义的Reducer要继承自己的父类

​ （2）Reducer的输入类型=Mapper的输出类型

​ （3）Reducer的业务逻辑在reduce()方法中

​ （4）ReduceTask进程对每一组相同k的<k,v>调用一次reduce方法

Driver:

​ 用于提交整个程序到Yarn集群，提交的是封装了MapReduce程序相关运行参数的job对象

在执行程序的时候，MR内部会对K进行排序

示例:

// Mapper
/*** 以WordCount案例为例* 自定义的Mapper类需要继承Hadoop提供的Mapper，* 并根据具体业务指定输入数据和输出数据的类型** 输入数据类型：KEYIN   读取数据的偏移量        VALUEIN  读取文件的一行数据* 输出数据类型：KEYOUT  输出数据的Key类型       VALUEOUT 输出数据的Value类型*/public class WordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {/*** map阶段的核心业务处理方法* 每输入一行数据会调用一次map方法* @param key      输入数据的key* @param value    输入数据的value* @param context  上下文对象* @throws IOException* @throws InterruptedException*/private Text outk = new Text();// 每个outv都是1private IntWritable outv=new IntWritable(1);@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//        获取当前输入的数据String line = value.toString();String[] data = line.split(" ");
//        遍历集合，封装 输出数据的key和valuefor (String datum : data) {outk.set(datum);context.write(outk,outv);}}
}

// Reducer
/**** Keyin    map端输出key的数据类型* valuein  map端输出value的数据类型* Keyout    reduce端输出key的数据类型* valueout  reduce端输出value的数据类型**/
public class WordCountReducer extends Reducer<Text, IntWritable,Text,IntWritable> {private Text outk = new Text();private IntWritable outv=new IntWritable();/*** reduce端核心业务处理方法* 一组相同key的values会调用一次reduce方法* @param key* @param values* @param context* @throws IOException* @throws InterruptedException*/@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int tol=0;
//      遍历valuesfor (IntWritable value : values) {
//          对value进行累加，最终输出结果tol+=value.get();}
//      封装k voutk.set(key);outv.set(tol);context.write(outk,outv);}
}

//Driver
/*** MR的驱动类，主要用于提交MR任务**/
public class WordCountDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//        声明配置对象Configuration configuration = new Configuration();
//        声明job对象Job job = Job.getInstance(configuration);
//        指定job的驱动类,获取到当前驱动类的实例job.setJarByClass(WordCountDriver.class);
//        指定当前job的mapper和reducerjob.setMapperClass(WordCountMapper.class);job.setReducerClass(WordCountReducer.class);
//        指定输入输出数据key和value的类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);
//        指定最终输出的key和value类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);
//        指定目录
//        FileInputFormat.setInputPaths(job,new Path("HdfsClient/src/others/wcinput/hello.txt"));
//        FileOutputFormat.setOutputPath(job, new Path("HdfsClient/src/others/wcoutput"));FileInputFormat.setInputPaths(job,new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));
//        提交job
//        true:让程序有个被监控的状态job.waitForCompletion(true);}
}

打包后传入服务器，运行hadoop jar MapReduce-1.0-SNAPSHOT.jar com.jojo.mr.WordCountDriver /wcinput /wcoutput1

com.jojo.mr.WordCountDriver是指执行的类路径【复制类名位置的reference获得】

MR执行流程

InputFormat --> Mapper --> Shuffle --> Reducer --> OutputFormat

​ map -> sort --------> copy->sort->reduce

InputFormat体系结构：

• FileInputFormat 是 InputFormat的子实现类，实现切片逻辑

​ 实现了getSplits()，负责切片

• TextInputFormat是FileInputFormat的子实现类。实现读取数据的逻辑

​ createRecordReader()返回一个RecordReader，在RecordReader中实现了读取数据的方式：按行读取。

• CombineFileInputFormat也是FileInputFormat的子实现类。此类中也实现了一套切片逻辑，适用于小文件计算场景。
  • 将输入目录下所有文件大小依次与设置的setMaxInputSplitSize比较，如果不大于设置的最大值，则逻辑上划分一个块，如果输入文件大于设置的两倍最大值则先以最大值切割一块，剩下一块再继续判断。如果不大于两倍最大值，则 将两个文件划分为两个虚拟存储块。
  • 切片过程，判断虚拟存储文件是否大于setMaxInputSplitSize值，大于等于则单独形成一个切片，不大于则与下一个虚拟存储文件进行合并，共同形成一个切片。
  • 一般来说，如果文件大小多次大于两倍setMaxInputSplitSize，则说明划分不合理。

【对InputFormat使用CTRL H 查看它的继承体系】

【指定InputFormat的具体实现】

job.setInputFormatClass(CombineFileInputFormat.class);
//设置虚拟存储切片最大值为4M
CombineFileInputFormat.setMaxInputSplitSize(job,4194304);

MapTask并行度

MapTask的并行度决定了Map阶段的任务处理并发度。

数据切片：数据切片只是在逻辑上对输入进行分片，不会在磁盘上对其进行切片。是MapReduce计算输入数据的单位，一个切片对应启动一个MapTask。一般情况下一个片的大小就等于一个块的大小

为什么片大小和切片大小默认是一致的：避免跨机器读取数据

对于切片的判断：当前文件的剩余大小/切片大小>1.1才继续切片，从而保证MapTask处理数据更加均衡。

Shuffle机制

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-AABainSU-1676377114693)(./shuffle.png)]

Map方法之后，Reduce方法之前的数据处理过程称之为Shuffle

Map:

• 内存环形缓冲区的大小为100MB，当数据写入了80MB时，会对数据进行溢写，留下的20MB继续进行写入操作

• 分区排序的算法使用的是快排

• 对多个分区进行合并的时候用的是归并排序

• Combiner为可选流程的意思是：可以选择在Map端提前对数据进行合并/压缩，缓解数据传到Reduce端时数据量过大的问题

Reduce:

• 有一个内存缓冲（优化手段），如果写入的数据量内存够装，就可以减少一次I/O操作

在整个shuffle过程中，分区、排序、Combiner是可控的，其他的是自动完成

Partitioner

​ Partitioner是Hadoop的分区器对象，负责给Map阶段输出数据选择分区的功能，默认实现HashPartitioner类

​ 按照输入key的hashCode值和ReduceTask数量进行取余，得到一个数字，把该数字作为当前k,v所属分区的编码，分区编码在Job提交的时候就根据指定ReduceTask的数量定义好了。

​ 对于源码：定位MapTask的map方法中 context.write(outk,outv)

​ write进入到 ChainMapContextImpl类的实现中

public void write(KEYOUT key, VALUEOUT value) throws IOException,  InterruptedException {output.write(key, value);}

​ output.write(key, value) 内部 NewOutputCollector

public void write(K key, V value) throws IOException, InterruptedException {collector.collect(key, value,partitioner.getPartition(key, value, partitions));}

​ partitioner.getPartition(key, value, partitions)；

​ 默认的分区规则实现 HashPartitioner类

public int getPartition(K key, V value,int numReduceTasks) {// 根据当前的key的hashCode值和ReduceTask的数量进行取余操作// 获取到的值就是当前kv所属的分区编号。return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;}

自定义分区器：

public class PhonePartitioner extends Partitioner<Text,FlowBean> {/*** 定义当前kv所属分区的规则* @param text* @param flowBean* @param numPartitions* @return*/public int getPartition(Text text, FlowBean flowBean, int numPartitions) {String phoneNum = text.toString();int phonePartitions;if (phoneNum.startsWith("136")){phonePartitions = 0;}else if (phoneNum.startsWith("137")){phonePartitions = 1;}else if (phoneNum.startsWith("138")){phonePartitions = 2;}else if (phoneNum.startsWith("139")){phonePartitions = 3;}elsephonePartitions = 4;return phonePartitions;}
}

在Driver中：

	 job.setPartitionerClass(PhonePartitioner.class);job.setNumReduceTasks(5);

ReduceTask数量>实际用的分区 时 会生成空的分区文件

​ < 报错

分区编号生成：根据指定的ReduceTask的数量从0开始，依次累加

WritableComparable

​ MapTask和ReduceTask均会对数据按照key进行排序。任何应用程序中的数据均会被排序，不管逻辑上是否需要。默认排序是安装字典顺序排序，实现该排序的方法是快排。

​ 对于MapTask，它会将处理的结果暂时放入环形缓冲区，缓冲区的使用率达到阈值后，会对缓冲区的数据进行一次快排，并将有序数据溢写到磁盘上，当数据处理完毕后，会对磁盘上所有文件进行归并排序。

​ 对于ReduceTask，会对每个MapTask上拷贝数据文件，文件大小超过一定阈值则溢写磁盘，否则存储于内存（即上面提到的减少I/O操作）。如果磁盘中文件数目达到一定阈值，进行一次归并生成更大的文件。所有数据拷贝完后，ReduceTask统一对内存和磁盘上所有数据进行一次归并排序。

排序分类
部分排序	MapReduce根据输入的key对数据集进行排序，保证输出文件内部有序。
全排序	最终输出结果只有一个文件，且文件内部有序。（只设置一个ReduceTask，该方法在处理大型文件时效率低，丧失了并行架构）
辅助排序（Grouping Comparator）	Reduce端对key进行分组，让一个或多个字段相同的key进入同一个Reduce方法。
二次排序	自定义排序过程中把compareTo判断条件设为两个

两种方式，对应java的CompareTo 和Comparator

WritableComparable

直接在比较的对象上实现WritableComparable接口的方式，Hadoop在运行的时候会自动生成比较器对象WritableComparator。

• 在Bean中，implements WritableComparable

• 重写compareTo方法

  public class FlowBean implements WritableComparable<FlowBean> {private long upFlow;private long downFlow;private long sumFlow;public FlowBean() {}public FlowBean(long upFlow, long downFlow) {this.upFlow = upFlow;this.downFlow = downFlow;this.sumFlow = upFlow+downFlow;}public void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(downFlow);out.writeLong(sumFlow);}public void set(long upFlow, long downFlow) {this.upFlow = upFlow;this.downFlow = downFlow;this.sumFlow = upFlow + downFlow;}public long getUpFlow() {return upFlow;}public void setUpFlow(long upFlow) {this.upFlow = upFlow;}public long getDownFlow() {return downFlow;}public void setDownFlow(long downFlow) {this.downFlow = downFlow;}public long getSumFlow() {return sumFlow;}public void setSumFlow() {this.sumFlow = upFlow+downFlow;}public void readFields(DataInput in) throws IOException {this.upFlow = in.readLong();this.downFlow = in.readLong();this.sumFlow = in.readLong();}@Overridepublic String toString() {return "FlowBean{" +"upFlow=" + upFlow +", downFlow=" + downFlow +", sumFlow=" + sumFlow +'}';}public int compareTo(FlowBean bean) {
  //        System.out.println(sumFlow+" and "+bean.getSumFlow());int result;// 倒序if (sumFlow > bean.getSumFlow()) {result = -1;}else if (sumFlow < bean.getSumFlow()) {result = 1;}else {result = 0;}return result;}
  }
  

  但是按照前面所述，需要注意只能对key进行排序。

WritableComparator

• 写一个Comparator比较器

• 比较器中需要指定当前比较器对象是为谁(Bean)服务，调用super进行关联

• 在Driver中指定自定义比较器对象

  public class FlowBeanComparator extends WritableComparator {//    指定当前比较器对象为谁服务public FlowBeanComparator() {super(FlowBean.class,true);}@Overridepublic int compare(WritableComparable a, WritableComparable b) {FlowBean abean=(FlowBean)a;FlowBean bbean=(FlowBean)b;int result;// 正序if (abean.getSumFlow() > bbean.getSumFlow()) {result = 1;}else if (abean.getSumFlow() < bbean.getSumFlow()) {result = -1;}else {result = 0;}return result;}
  }
  

  // Driver
  //        设置自定义的比较器对象job.setSortComparatorClass(FlowBeanComparator.class);
  

  当程序中既重写了compareTo方法，又有一个新的比较器时，效果是比较器的效果。

  但是因为Comparator的实现依赖于WritableComparable方法，所以Bean必须实现WritableComparable接口。

Hadoop获取比较器对象的规则：

​ job提交的时候获取当前MR输出数据key的比较器对象–>能通过配置获取到指定的比较器对象的class则直接通过反射实例化

​ -->通过配置没获取到指定的比较器对象则接着判断当前参与比较的对象是否实现了WritableComparable接口。如果是Hadoop自身的数据类型，则获取自身类型的比较器对象，且会重新再加载一遍，防止极端情况发生GC垃圾回收，比较器没了。如果还是没获取到，则说明是自定义的，并为当前参与比较的对象生成比较器。

Hadoop自身数据类型如何拥有比较器对象：

​ 自身实现了WritableComparable接口，且定义了自己的比较器对象。此外，类中还含有一个静态代码块。类的比较器对象被管理到一个Map中，以当前类的class文件为key，当前类的比较器为value。

Combiner

使用场景

• 为了提升MR程序的运行效率，减轻ReduceTask的压力，减少磁盘I/O开销。
• 意义在于对每个MT的输出进行局部汇总，以减小网络传输量。
• Combiner继承Reducer。两者的区别在于运行位置，Combiner在每个MapTask所在节点运行，Reducer接收全局所有Mapper的输出结果。
• Combiner发生在Map阶段
• Combiner不是所有场景都能用，Reduce端处理的数据考虑到多个MapTask的数据的整体集时，不能使用（如求平均数）。

使用方法

• 自定义一个Combiner类，继承Hadoop提供的Reducer
• 在job中指定自定义的Combiner类

public class WordCombiner extends Reducer <Text, IntWritable,Text,IntWritable> {private Text outk=new Text();private IntWritable outv=new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int tol=0;for (IntWritable value : values) {tol+=value.get();}outk.set(key);outv.set(tol);context.write(outk,outv);}
}

//          指定自定义的Combiner类job.setCombinerClass(WordCombiner.class);

OutputFormat

定义

• OutputFormat是MapReduce输出的基类，所有实现MapReduce输出都实现了OutputFormat的接口
• OutputFormat主要负责最终数据的输出
• OutputFormat类的体系结构
• FileOutputFormat是OutputFormat的子类（实现类），对 checkOutputSpecs() 做了具体的实现
• TextOutputFormat是FileOutputFormat的子类，对 getRecordWriter 做了具体实现
• 使用场景：对MR最终结果有个性化制定的需求，就可以自定义OutputFormat。

自定义OutputFormat

场景：将文件中包含sina的网址和不包含sina的网址输出至不同的文件夹

// Mapper:
// 除了将不同网址分离，没有别的需求，所以把网址作为key，value为空
public class FilterMapper extends Mapper<LongWritable, Text,Text, NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {context.write(value,NullWritable.get());}
}

// Reducer:
public class FilterReducer extends Reducer<Text, NullWritable,Text,NullWritable> {@Overrideprotected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {for (NullWritable value : values) {context.write(key,NullWritable.get());}}
}

// Driver
public class FilterDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {Configuration configuration = new Configuration();Job job = Job.getInstance(configuration);job.setReducerClass(FilterReducer.class);job.setMapperClass(FilterMapper.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);job.setMapOutputValueClass(NullWritable.class);job.setMapOutputKeyClass(Text.class);job.setJarByClass(FilterDriver.class);
//	============================
//      除了常规设置外，由于要自定义OutputFormat，因此在此处指定自定义方式job.setOutputFormatClass(FilterOutputFormat.class);FileInputFormat.setInputPaths(job,new Path("../third/MapReduce/src/others/log"));FileOutputFormat.setOutputPath(job,new Path("../third/MapReduce/src/others/log/output"));job.waitForCompletion(true);}
}

// 自定义的格式必须继承FileOutputFormat
public class FilterOutputFormat extends FileOutputFormat {public RecordWriter getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
// 使用自己定义的RecordWriterLogRecordWriter logRecordWriter = new LogRecordWriter(job);return logRecordWriter;}
}

//自定义的RecordWriter  需要继承原有的RecordWriter
public class LogRecordWriter extends RecordWriter {
// 定义在外部方便closeprivate FSDataOutputStream oneOut;private FSDataOutputStream secondOut;public LogRecordWriter(TaskAttemptContext job) throws IOException {//在构造器中获取传过来的job中的配置，Hadoop文件系统对象FileSystem fileSystem = FileSystem.get(job.getConfiguration());oneOut=fileSystem.create(new Path("../third/MapReduce/src/others/log/oneout.txt"));secondOut=fileSystem.create(new Path("../third/MapReduce/src/others/log/secondout.txt"));}public void write(Object key, Object value) throws IOException, InterruptedException {String s = key.toString();if (s.contains("sina"))oneOut.writeBytes(s+"\n");elsesecondOut.writeBytes(s+"\n");}public void close(TaskAttemptContext context) throws IOException, InterruptedException {// 使用hadoop的工具类关闭文件IOUtils.closeStream(oneOut);IOUtils.closeStream(secondOut);}
}

join

Reduce Join

• 先在map端对数据进行收集，找到对应的key和value，到Reduce端对应关联。

// 两个表：订单表记录：订单号 商品编号 数量
//        商品表记录： 商品编号 商品名字// 合并输出文件：订单号，商品编号，数量和对应的商品名字
/**
1001 01 1
1002 02 2
1003 03 3
1004 01 4
1005 02 5
1006 03 601 小米
02 华为
03 格力
*/

商品实物类

public class ProductBean implements Writable {private String id;   // idprivate Integer amount;  // amountprivate String pname; // pnameprivate String pid;  // pidprivate String flag; // 来源，是商品表还是订单表@Overridepublic String toString() {return "ProductBean{" +"id='" + id + '\'' +", amount=" + amount +", pname='" + pname + '\'' +", pid='" + pid + '\'' +'}';}public void write(DataOutput out) throws IOException {out.writeUTF(id);out.writeUTF(pid);out.writeInt(amount);out.writeUTF(pname);out.writeUTF(flag);}public void readFields(DataInput in) throws IOException {id=in.readUTF();pid=in.readUTF();amount=in.readInt();pname=in.readUTF();flag=in.readUTF();}public ProductBean() {}public String getId() {return id;}public void setId(String id) {this.id = id;}public Integer getAmount() {return amount;}public void setAmount(Integer amount) {this.amount = amount;}public String getPname() {return pname;}public void setPname(String pname) {this.pname = pname;}public String getPid() {return pid;}public void setPid(String pid) {this.pid = pid;}public String getFlag() {return flag;}public void setFlag(String flag) {this.flag = flag;}
}

Mapper

public class ReduceJoinMapper extends Mapper<LongWritable, Text, Text,ProductBean> {private Text outk=new Text();private ProductBean outv=new ProductBean();private String filename;@Overrideprotected void setup(Context context) throws IOException, InterruptedException {InputSplit inputSplit = context.getInputSplit();FileSplit fileSplit = (FileSplit)inputSplit;filename  = fileSplit.getPath().getName();}@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString();if (filename.contains("order")){String[] split = line.split(" ");outk.set(split[1]);outv.setId(split[0]);outv.setAmount(Integer.parseInt(split[2]));outv.setPname("");outv.setFlag("order");outv.setPid(split[1]);}else {String[] split = line.split(" ");outk.set(split[0]);outv.setId("");outv.setPid(split[0]);outv.setAmount(0);outv.setFlag("pd");outv.setPname(split[1]);}context.write(outk,outv);}
}

Reducer

public class ReduceJoinReducer extends Reducer<Text, ProductBean, ProductBean, NullWritable> {private ArrayList<ProductBean> productBeans = new ArrayList<ProductBean>();private ProductBean brand = new ProductBean();@Overrideprotected void reduce(Text key, Iterable<ProductBean> values, Context context) throws IOException, InterruptedException {for (ProductBean value : values) {if (value.getFlag().equals("order")) {
//             由于集合里面存的是指针对象，而遍历的values每个value也是指向的某一个对象
//             每次遍历移动的是指针，所以不能直接使用add对value进行添加，而是创建一个实体
//             否则添加的永远是最后指向的那个对象try {ProductBean thisProductBean = new ProductBean();BeanUtils.copyProperties(thisProductBean, value);productBeans.add(thisProductBean);} catch (IllegalAccessException e) {e.printStackTrace();} catch (InvocationTargetException e) {e.printStackTrace();}} else {try {BeanUtils.copyProperties(brand, value);} catch (IllegalAccessException e) {e.printStackTrace();} catch (InvocationTargetException e) {e.printStackTrace();}}}for (ProductBean bean : productBeans) {bean.setPname(brand.getPname());context.write(bean, NullWritable.get());}productBeans.clear();}
}

弊端

• 在数据量太大的情况下不适合（Redis，Mongodb）

• 合并操作在Reduce阶段完成，Reduce处理压力太大

• Map运算负载低，资源利用率不高

• Reduce阶段极易产生数据倾斜(分类后假如有3个分区，可能出现三个分区需要处理的数据量差异大)

Map Join

• Map Join的思路：当Map Task执行的时候，先把数据量较小的文件缓存到内存当中。MapTask正常将数据量大的文件按行读入，每处理一行数据，就根据文件中的key到内存中与数据量较小的文件进行join。

• Map join 适用于一个表非常小，一个表非常大的场景

Mapper

/*** 处理缓存文件：将job中设置的缓存路径获取到* 根据缓存路径再结合输入流把内容写入到内存的容器中维护**/
public class MapJoinMapper extends Mapper<LongWritable, Text,Text, NullWritable> {private HashMap<String,String>pdMap=new HashMap<String, String>();private Text outk =  new Text();@Overrideprotected void setup(Context context) throws IOException, InterruptedException {
//        将job中设置的缓存路径获取到URI[] cacheFiles = context.getCacheFiles();URI cacheFile = cacheFiles[0];
//        准备输入流对象FileSystem fileSystem = FileSystem.get(context.getConfiguration());FSDataInputStream pdInput = fileSystem.open(new Path(cacheFile));
//        通过流对象将数据读入，保存到内存中BufferedReader reader = new BufferedReader(new InputStreamReader(pdInput, "UTF-8"));
//        按行读取String line;while ((line=reader.readLine())!=null){
//            将数据保存到map中String[] data = line.split(" ");pdMap.put(data[0],data[1]);}IOUtils.closeStream(pdInput);}@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//        获取当前行数据String line = value.toString();String[] orderData = line.split(" ");
//        关联String pname=pdMap.get(orderData[1]);String result = orderData[0]+" "+pname+" "+orderData[2];outk.set(result);context.write(outk,NullWritable.get());}
}

Driver

public class MapJoinDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//        声明配置对象Configuration configuration = new Configuration();
//        声明job对象Job job = Job.getInstance(configuration);
//        指定job的驱动类,获取到当前驱动类的实例job.setJarByClass(MapJoinDriver.class);
//        指定当前job的mapper和reducerjob.setMapperClass(MapJoinMapper.class);
//        指定输入输出数据key和value的类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);
//        设置Reduce数量为0,因为默认为1job.setNumReduceTasks(0);
//      设置缓存文件的路径file:///E:/Code/javaCodejob.addCacheFile(URI.create("file:///E:/Code/javaCode/third/MapReduce/src/others/joinPhone/pd.txt"));
//        指定目录FileInputFormat.setInputPaths(job,new Path("file:///E:/Code/javaCode//third/MapReduce/src/others/map"));FileOutputFormat.setOutputPath(job, new Path("file:///E:/Code/javaCode/third/MapReduce/src/others/MapJoinPhoneOutput"));//        提交job
//        true:让程序有个被监控的状态job.waitForCompletion(true);}
}

计数器

setup(//        计数器context.getCounter("MapTask","setup").increment(1);
)map(//        计数器context.getCounter("MapTask","map").increment(1);
)/**
控制台打印：	
MapTaskmap=6setup=1
*/

数据清洗ETL

​ 在运行MapReduce之前，对数据进行清洗，清理掉不符合要求的数据。

​

Yarn

Yarn是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台，而MapReduce等运算程序则相当于运行于操作系统之上的应用程序

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-3KPwdbV5-1676377114693)(F:\md\大数据\yarn.png)]

• 主要由ResourceManager、NodeManager、ApplicationMaster和Container等组件构成
• Hadoop作业调度器主要有三种：FIFO、Capacity Scheduler和Fair Scheduler
  • FIFO对于批处理作业、交互式作业、生产性作业的资源调度不灵活
  • Capacity Scheduler容量调度器：支持多个队列，每个队列配置一定的资源量，每个队列采用FIFO调度策略。每个队列可设定一定比例的资源最低保证和使用上限，且每个用户也可设定一定的资源使用上限以防止资源滥用，当一个队列的资源有剩余时，可暂时将剩余资源共享给其他队列。
  • Fair Scheduler公平调度器：支持多队列多作业，同队列所有任务共享资源，在时间尺度上获得公平的资源。让所有的作业随着时间的推移，都能平均地获取等同的共享资源。

压缩

压缩用于减少底层存储系统（HDFS）读写字节数，对于节省资源、最小化磁盘I/O和网络传输有利，可以在任意MR阶段启用。

通过对Mapper、Reducer运行过程的数据进行压缩，以减少磁盘I/O，但增加了CPU运算负担。

使用原则

（1）运算密集型job，少用压缩

（2）IO密集型job，多用压缩

类型

压缩格式	hadoop自带？	算法	文件扩展名	是否可切分	换成压缩格式后，原来的程序是否需要修改
DEFLATE	是，直接使用	DEFLATE	.deflate	否	和文本处理一样，不需要修改
Gzip	是，直接使用	DEFLATE	.gz	否	和文本处理一样，不需要修改
bzip2	是，直接使用	bzip2	.bz2	是	和文本处理一样，不需要修改
LZO	否，需要安装	LZO	.lzo	是	需要建索引，还需要指定输入格式
Snappy	是，直接使用	Snappy	.snappy	否	和文本处理一样，不需要修改

名字	优点	缺点	应用场景
Gzip	压缩率较高，压缩/解压速度快，Hadoop自带，使用方便	不支持Split	每个文件压缩后在130MB以内都可以考虑用Gzip
Bzip2	支持Split，压缩率比Gzip高，Hadoop自带	压缩/解压速度慢	对速度要求不高，但需要较高压缩率的情况；或输出后数据较大，需要将数据压缩存档，且以后数据使用较少的情况；或单个大文本文件想压缩，且需要Split的情况。
Lzo	合理的压缩率，压缩/解压速度较快，支持Split	压缩率比Gzip低一些，Hadoop本身不支持，需要安装，为至此Split，需要新建索引	大文本文件，且压缩后大于200MB以上的场景适用，且单个文件越大，Lzo优点越明显。
Snappy	高压缩速度，合理的压缩率	不支持Split，压缩率比Gzip低	MapReduce作业中Map输出数据较大的时候，作为Map到Reduce的中间数据压缩格式；或作为一个MapReduce作业的输出和另一个MapReduce作业的输入。

在Map阶段进行压缩：

注意：configuration的设置应该在job对象声明的前面，否则后面加入的配置不能生效。

//        声明配置对象Configuration configuration = new Configuration();
//        设置在Mapper阶段进行压缩
//        两种设置方式
//        configuration.set("mapreduce.map.output.compress","true");configuration.setBoolean("mapreduce.map.output.compress",true);
//      设置编解码器configuration.set("mapreduce.map.output.compress.codec","org.apache.hadoop.io.compress.DefaultCodec");
//        声明job对象Job job = Job.getInstance(configuration);。。。

在Reducer端进行压缩

//        设置在Reducer端输出的时候压缩configuration.set("mapreduce.output.fileoutputformat.compress","true");configuration.set("mapreduce.output.fileoutputformat.compress.codec","org.apache.hadoop.io.compress.DefaultCodec");

文件压缩和解压缩

压缩：

public class CompressTest {/*** 对文件进行压缩*/String srcFile="E:\\Code\\javaCode\\third\\MapReduce\\src\\others\\jianai\\ja.txt";String destFile="E:\\Code\\javaCode\\third\\MapReduce\\src\\others\\jianai\\ja";/*** 压缩:通过一个能够具备压缩功能输出流将文件写出* 不会写怎么办：* 代码参考：InputFormat中切片方法getSplits的实现类：FileInputFormat，其中使用了isSplitable方法，再跟进去找到具体实现类：TextInputFormat中的isSplitable，其中，通过工厂的方式声明了编码器。点进Factory里面可以发现，实际是使用了反射*/@Testpublic void testCompress() throws IOException, ClassNotFoundException {// 声明一个输入流FileInputStream fileInputStream=new FileInputStream(new File(srcFile));// 获取一个编解码器（压缩工具对象）Configuration conf = new Configuration();String classPath = "org.apache.hadoop.io.compress.DefaultCodec";Class<?> codecClass = Class.forName(classPath);// 根据分析，从源码中copy了这段代码，可以发现缺少class和conf，1.编码器的类路径是固定的，2.声明一个conf// 怎么找的强转类型：也是从源码中抄的CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);// 声明一个输出流FileOutputStream fileOutputStream=new FileOutputStream(new File(destFile)+codec.getDefaultExtension());// 把普通的输出流让CompressionCodec包装一下CompressionOutputStream outputStream = codec.createOutputStream(fileOutputStream);// 读写数据IOUtils.copyBytes(fileInputStream,outputStream,conf);// 关闭流IOUtils.closeStream(fileInputStream);IOUtils.closeStream(outputStream);}
}

解压缩

public class CompressTest {/*** 对文件进行解压缩*/String srcFile="E:\\Code\\javaCode\\third\\MapReduce\\src\\others\\jianai\\ja.deflate";String destFile="E:\\Code\\javaCode\\third\\MapReduce\\src\\others\\jianai\\ja.txt";/*** 解压缩：通过一个能够具备解压缩功能输入流将文件写出，步骤和压缩是反过来的，不同的是，压缩中的编码格式是人为指定的，而解压缩的解码方式是通过文件名自动获取的。*/@Testpublic void testDeCompress() throws IOException, ClassNotFoundException {// 声明一个输入流FileInputStream in = new FileInputStream(new File(srcFile));// 获取一个编解码器（解压缩工具对象）Configuration conf = new Configuration();CompressionCodec codec =new CompressionCodecFactory(conf).getCodec(new Path(srcFile));// 把普通的输入流让CompressCodec包装一下CompressionInputStream inputStream = codec.createInputStream(in);// 声明输出流FileOutputStream fileOutputStream = new FileOutputStream(new File(destFile));// 读写数据IOUtils.copyBytes(inputStream,fileOutputStream,conf);// 关流IOUtils.closeStream(inputStream);IOUtils.closeStream(fileOutputStream);}
}

Zookeeper

Zookeeper是基于观察者设计模式的分布式服务管理框架

Zookeeper = 文件系统 + 通知机制

特点

（1） Zookeeper由一个leader和多个follower组成

（2） 集群中有半数以上节点存活，Zookeeper就可以正常服务

（3） 全局数据一致，每个Server保存一份相同的数据副本，Client无论连接到哪个Server，数据都一致

（4） 更新请求顺序进行，来自一个Client的更新请求按其发送顺序执行

（5） 原子性，数据更新时，一次数据更新要么成功要么失败

（6） 实时性，在一定时间范围内，Client能读到最新数据（毫秒级）

存储

整体可以看作一棵树，每个节点默认能存储1MB数据，每个ZNode都可以通过路径唯一访问。Zookeeper中没有文件的概念，节点下直接存储内容。

HA高可用

现有集群存在哪些问题

1. HDFS集群 单个NN场景下

   问题：如果NN故障了，整个HDFS集群就不可用了（中心化集群的特点）

   解决方案：配置多个NN

   ​ 配置多个NN的场景下，由哪一台对外进行服务呢？

   ​ 当HDFS实现多NN的高可用后，但是只有一台NN对外提供服务，其他的NN都是替补，当正在提供服务的NN宕机故障了，其他的NN自动切换为Active状态。

   ​ 当一台NN故障后，其他NN如何争抢上位？

   ​ 采用高可用集群中的自动故障转移机制来完成切换

   ​ 在高可用集群中还需要2NN吗？

   ​ 不需要2NN，元数据的维护策略还继续保持原样，但是在高可用集群中，会添加一个新的服务（journalNode：本身自己也要搭建成一个集群的状态，他和Zookeeper集群很像，存活机器数量过半，就能正常提供服务。它主要负责编辑日志文件的内容共享）

2. Yarn集群下

   也存在RM单点故障的问题

工作机制

• ZooKeeper是维护少量协调数据，通知客户端这些数据的改变和监视客户端故障的高可用服务；

• ZKFC是自动故障转移中的另一个新组件，是ZooKeeper的客户端，也监视和管理NameNode的状态。每个运行NameNode的主机都运行了一个ZKFC进程；

  • ZKFC负责健康检测，定期ping监测的NN，NN及时回复即判断为健康节点
  • 保持与Zookeeper的会话，如果本地NN处于active状态，则额外保持一个znode锁
  • 如果ZKFC发现没有其他节点持有znode，则为自己获取znode，且负责的NN转为Active

怎么防止脑裂？

​ 1）ssh发送kill指令（要上位的节点ssh登录到要代替的节点上，使用kill命令杀死namenode，防止原NN是假死状态）

​ 2）调用用户自定义脚本程序（如果ssh补刀还没成功，就调用用户自定义脚本继续补刀）

HDFS-HA

• 通过多个NameNode消除单点故障

• Edits日志只有Active状态的NameNode节点可以做写操作，所有的NN都可以读取Edits

• 利用qjournal和NFS实现共享Edits放在一个共享存储中管理

• 每个NN都有一个zkfailover负责监控，利用zk进行状态标识，切换状态时，由zkfailover负责切换（要避免脑裂现象）

• 必须保证两个NN之间能ssh免密登录

• 保证隔离，同一时刻仅有一个NN对外提供服务

搭建步骤注意事项

• zoo.cfg中的Server.X X这个数字要与myid中的数字匹配

• 格式化nn之前，要删除hadoop目录下的data和log目录（避免mycluster产生的id与原id不匹配）

  hdfs namenode -format

• core-site.xml,hdfs-site.xml,yarn-site.xml都需要改内容

• 启动zk后，要初始化HA在zk中的状态

  hdfs zkfc -formatZK

HDFS联邦架构

当集群中数据量超级大时，NameNode的内存成了性能的瓶颈，提出了联邦机制

将NameNode划分成不同的命名空间并进行编号。不同的命名空间之间相互隔离互不干扰，在DataNode中创建目录，此目录对应命名空间的编号，编号相同的数据由对应的命名空间进行管理

但存储需求也会暴增
​ 当一台NN故障后，其他NN如何争抢上位？

​ 采用高可用集群中的自动故障转移机制来完成切换

​ 在高可用集群中还需要2NN吗？

​ 不需要2NN，元数据的维护策略还继续保持原样，但是在高可用集群中，会添加一个新的服务（journalNode：本身自己也要搭建成一个集群的状态，他和Zookeeper集群很像，存活机器数量过半，就能正常提供服务。它主要负责编辑日志文件的内容共享）

2. Yarn集群下

   也存在RM单点故障的问题

工作机制

• ZooKeeper是维护少量协调数据，通知客户端这些数据的改变和监视客户端故障的高可用服务；

• ZKFC是自动故障转移中的另一个新组件，是ZooKeeper的客户端，也监视和管理NameNode的状态。每个运行NameNode的主机都运行了一个ZKFC进程；

  • ZKFC负责健康检测，定期ping监测的NN，NN及时回复即判断为健康节点
  • 保持与Zookeeper的会话，如果本地NN处于active状态，则额外保持一个znode锁
  • 如果ZKFC发现没有其他节点持有znode，则为自己获取znode，且负责的NN转为Active

怎么防止脑裂？

​ 1）ssh发送kill指令（要上位的节点ssh登录到要代替的节点上，使用kill命令杀死namenode，防止原NN是假死状态）

​ 2）调用用户自定义脚本程序（如果ssh补刀还没成功，就调用用户自定义脚本继续补刀）

HDFS-HA

• 通过多个NameNode消除单点故障

• Edits日志只有Active状态的NameNode节点可以做写操作，所有的NN都可以读取Edits

• 利用qjournal和NFS实现共享Edits放在一个共享存储中管理

• 每个NN都有一个zkfailover负责监控，利用zk进行状态标识，切换状态时，由zkfailover负责切换（要避免脑裂现象）

• 必须保证两个NN之间能ssh免密登录

• 保证隔离，同一时刻仅有一个NN对外提供服务

搭建步骤注意事项

• zoo.cfg中的Server.X X这个数字要与myid中的数字匹配

• 格式化nn之前，要删除hadoop目录下的data和log目录（避免mycluster产生的id与原id不匹配）

  hdfs namenode -format

• core-site.xml,hdfs-site.xml,yarn-site.xml都需要改内容

• 启动zk后，要初始化HA在zk中的状态

  hdfs zkfc -formatZK

HDFS联邦架构

当集群中数据量超级大时，NameNode的内存成了性能的瓶颈，提出了联邦机制

将NameNode划分成不同的命名空间并进行编号。不同的命名空间之间相互隔离互不干扰，在DataNode中创建目录，此目录对应命名空间的编号，编号相同的数据由对应的命名空间进行管理

但存储需求也会暴增