河源市东源县建设局网站,有没有做logo的网站,小程序商店助手,做标签网站Hadoop-MapReduce 文章目录Hadoop-MapReduce1 MapRedcue的介绍1.1 MapReduce定义1.2 MapReduce的思想1.3MapReduce优点1.4MapReduce的缺点1.5 MapReduce进程1.6 MapReduce-WordCount1.6.1 job的讲解2 Hadoop序列化2.1 序列化的定义2.2 hadoop序列化和java序列化的区别3 MapRedu…Hadoop-MapReduce 文章目录Hadoop-MapReduce1 MapRedcue的介绍1.1 MapReduce定义1.2 MapReduce的思想1.3MapReduce优点1.4MapReduce的缺点1.5 MapReduce进程1.6 MapReduce-WordCount1.6.1 job的讲解2 Hadoop序列化2.1 序列化的定义2.2 hadoop序列化和java序列化的区别3 MapReduce 的原理3.1 MapReduce 工作的过程3.2 InputFormat 数据输入3.2.1 切片3.2.2 FileInputFormat3.2.3 TextInputFormat3.2.4 CombineTextInputFormat3.3 MapReduce工作机制3.3.1 MapTask工作机制3.3.2 Partition分区3.3.3 Combiner合并3.3.4 ReduceTask工作机制3.3.5 shuffle机制3.3.6 排序的解释4 数据压缩4.1 压缩算法对比4.2 压缩位置选择1 MapRedcue的介绍
1.1 MapReduce定义
MapReduce是一个进行分布式运算的编程框架使用户开发基于hadoop进行数据分析的核心框架。 MapReduce 核心功能就是将用户编写的业务逻辑代码和自带的默认组件整合成一个完整的 分布式运算程序并发运行在一个 Hadoop 集群上。
1.2 MapReduce的思想
MapReduce的思想核心是分而治之适用于大规模数据处理场景。
map负责分将复杂的任务拆解成可以并行计算的若干个任务来处理
reduce负责合对map阶段的结果进行全局汇总
比如说老师作业留的有点多一个人写太费劲了就可以用MapReduce这种分而治之的思想将作业进行map处理分给不同的人最后所有写完的部分发到群里进行reduce汇总复杂的作业简简单单。
1.3MapReduce优点
易于编程 MapReduce将做什么和怎么做分开了提供了一些接口程序员只需关注应用层上的问题。具体如何实现并行计算任务则被隐藏了起来。
扩展性 当计算资源不足时可以增加机器来提高扩展能力
高容错 一台机器挂了可以将计算任务转移到另一台节点上进行
适合PB级海量数据的离线处理
1.4MapReduce的缺点
不擅长实时计算 无法做到在毫秒级别返回结果
不擅长流式计算 MapReduce处理的数据源只能是静态的不能动态变化
不擅长DAG(有向无环图)计算 每个MR作业处理结束结果都会写入到磁盘造成大量的磁盘IO导致性能低下
1.5 MapReduce进程
一个MapReduce程序在分布式运行时有三类的实例进程
MrAppMaster : 负责整个程序的过程调度及状态协调MapTask : 负责Map阶段的数据处理流程ReduceTask 负责Reduce阶段的数据处理流程
1.6 MapReduce-WordCount
import java.io.IOException;
import java.util.StringTokenizer;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class WordCount {public static class TokenizerMapperextends MapperObject, Text, Text, IntWritable{private final static IntWritable one new IntWritable(1);private Text word new Text();public void map(Object key, Text value, Context context) throws IOException, InterruptedException {StringTokenizer itr new StringTokenizer(value.toString());while (itr.hasMoreTokens()) {word.set(itr.nextToken());context.write(word, one);}}}public static class IntSumReducerextends ReducerText,IntWritable,Text,IntWritable {private IntWritable result new IntWritable();public void reduce(Text key, IterableIntWritable values,Context context) throws IOException, InterruptedException {int sum 0;for (IntWritable val : values) {sum val.get();}result.set(sum);context.write(key, result);}}public static void main(String[] args) throws Exception {Configuration conf new Configuration();Job job Job.getInstance(conf, word count);job.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);job.setCombinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));System.exit(job.waitForCompletion(true) ? 0 : 1);}
}以上代码实现了两个类TokenizerMapper和IntSumReducer它们分别实现了Map和Reduce功能。
Map函数将输入的每一行文本进行分词并将每个单词映射为一个键值对其中键为单词值为1然后将这些键值对输出给Reduce函数。
Reduce函数将相同键的值相加并将最终结果输出。
在这个例子中CombinerClass被设置为相同的Reducer类用于在Map任务结束后本地合并中间结果以减少网络传输。
最后将输入文件和输出文件的路径作为命令行参数传递给main函数并启动MapReduce作业。
1.6.1 job的讲解
在Hadoop MapReduce程序中Job对象是用来定义和运行一个MapReduce作业的。
Job对象的主要功能是封装了整个MapReduce作业的配置和运行信息包括输入数据和输出数据的路径、Mapper类和Reducer类的设置、中间结果的输出类型和格式、作业的提交方式等。
在main函数中我们创建一个Job对象并设置它的相关属性。
Job.getInstance()方法返回一个新的Job实例其中的Configuration对象用来指定作业的一些配置信息。
setJarByClass()方法用来设置作业的jar包它的参数是定义MapReduce作业的主类。
setMapperClass()、setCombinerClass()和setReducerClass()方法用来指定Mapper、Combiner和Reducer的实现类。
setOutputKeyClass()和setOutputValueClass()方法分别用来设置MapReduce作业的输出键和输出值的类型。
FileInputFormat.addInputPath()和FileOutputFormat.setOutputPath()方法用来指定输入文件和输出文件的路径。
最后我们调用job.waitForCompletion()方法来提交并运行作业并等待作业完成。
如果作业成功完成waitForCompletion()方法将返回true否则返回false。
如果作业失败我们可以通过job.getJobState()方法来获取作业的状态信息或者查看作业的日志信息来进行排错和调试。
2 Hadoop序列化
2.1 序列化的定义
序列化就是将内存中对象转换成字节序列便于存储到磁盘和网络传输
反序列化时将字节序列或磁盘中的持久化数据转换成内存中的对象
一般来说对象只能在本地进程中使用不能通过网络发送到另一台计算机
序列化可以存储对象可以将对象发送到远程计算机
2.2 hadoop序列化和java序列化的区别
Hadoop序列化和Java序列化都是将对象转换为字节序列以便于在网络上传输或者存储到磁盘等持久化存储介质中。它们的主要区别在于以下几点 序列化速度和效率不同Hadoop序列化比Java序列化更快因为它采用的是二进制格式而Java序列化采用的是基于文本的XML或JSON格式。 支持的数据类型不同Hadoop序列化支持的数据类型比Java序列化更多包括基本类型、数组、集合、映射、枚举、自定义类等。 序列化后的数据大小不同Hadoop序列化生成的字节流比Java序列化生成的字节流更小因为它使用更紧凑的二进制格式这对于在网络上传输和存储到磁盘等介质中非常重要。 可移植性不同Java序列化生成的字节流只能被Java程序读取而Hadoop序列化生成的字节流可以被任何语言的程序读取因为它使用了通用的二进制格式。
总的来说Hadoop序列化更适合用于大规模数据的处理和分布式计算而Java序列化更适合用于小规模数据的传输和存储。
3 MapReduce 的原理
3.1 MapReduce 工作的过程 数据输入MapReduce从Hadoop分布式文件系统HDFS中读取输入数据并将其分成固定大小的数据块每个数据块大小通常为64MB。
Map阶段在Map阶段MapReduce将每个数据块分发给一组可扩展的计算节点每个计算节点运行Map函数来处理它们分配的数据块。Map函数将输入数据转换为一组键值对Key-Value Pairs的形式这些键值对可以被后续的Reduce函数处理。
Shuffle阶段在Map函数处理完数据之后MapReduce框架将所有的键值对按照它们的键进行排序并将相同键的值合并在一起。这个过程通常被称为“Shuffle”。
Reduce阶段在Reduce阶段MapReduce框架将合并后的键值对发送到一组可扩展的计算节点。每个节点运行Reduce函数来处理它们收到的所有键值对并生成最终的输出结果。
数据输出在Reduce函数处理完数据之后MapReduce将输出结果写入HDFS中。
这些步骤中Map和Reduce函数是由开发者自行编写的它们实现了具体的业务逻辑。MapReduce框架提供了分布式计算的基础设施负责管理计算节点、任务分配、故障处理等任务以保证整个计算过程的可靠性和高效性。
总的来说MapReduce框架的原理是将大数据集划分成多个小数据块然后将这些数据块分发给多个计算节点并行处理最后将处理结果合并为一个最终结果。它通过这种方式来充分利用集群中的计算资源提高计算效率和数据处理能力。
3.2 InputFormat 数据输入
3.2.1 切片
数据块Block时HDFS在物理上对数据进行切块是HDFS存储数据的单位
数据切片数据切片是在逻辑上对输入进行切片。切片是MR程序计算输入数据的单位一个切片会启动一个MapTask
客户端提交job时的切片数决定了map阶段的并行度
默认情况下切片大小为BlockSize
切片不会考虑数据整体是对每个文件进行单独切片
3.2.2 FileInputFormat
在MapReduce中FileInputFormat是一个抽象类用于定义如何将文件分割成输入数据块并生成适合Mapper处理的RecordReader。它是MapReduce中的输入格式类之一用于读取Hadoop分布式文件系统HDFS或本地文件系统中的数据。
FileInputFormat包括两个关键方法getSplits()和createRecordReader()。
getSplits()方法将输入文件划分成适合Map任务的数据块每个数据块对应一个Map任务。该方法返回一个InputSplit对象的数组其中每个InputSplit表示一个文件数据块。
createRecordReader()方法创建一个RecordReader对象用于读取InputSplit中的数据块。RecordReader负责读取一个数据块中的所有记录并将它们转换成key-value对。
FileInputFormat还提供了一些其他的方法如isSplitable()用于判断一个文件是否可以被划分成多个数据块。
Hadoop提供了一些预定义的FileInputFormat类如TextInputFormat用于读取文本文件SequenceFileInputFormat用于读取SequenceFile格式的文件等用户也可以通过继承FileInputFormat自定义输入格式类。
3.2.3 TextInputFormat
TextInputFormat是FileInputFormat默认的实现类按行读取每条记录
key为该行的起始字节偏移量为LongWritable类型
value 为这一行的内容不包括终止符为Text类型
3.2.4 CombineTextInputFormat
TextInputFormat是按文件进行规划分片不管文件有多小都是是一个单独的切片这样会产生大量的MapTask效率低下
CombineTextInputFormat用于小文件过多的场景可以将多个小文件在逻辑上划分到一个切片
决定哪些块放入同一个分片时CombineTextInputFormat会考虑到节点和机架的因素所以在MR作业处理输入的速度不会下降
CombineTextInputFormat不仅可以很好的处理小文件在处理大文件时也有好处因为它在每个节点生成了一个分片分片可能又多个块组成CombineTextInputFormat使map操作中处理的数据量和HDFS中文件块的大小的耦合度降低了
3.3 MapReduce工作机制
3.3.1 MapTask工作机制 读取输入数据MapTask通过InputFormat获得RecordReader从输入InputSplit中解析出KVMap阶段将解析出的KV交给map()函数处理产生一系列新的KVCollect收集数据处理完成之后会调用OutputCollector.collect()输出结果。在该函数的内部会生成KV分区写入环形缓冲区中Spill阶段环形缓冲区满了之后MR会将数据写到本地磁盘形成一个临时文件在写入之前会对数据进行一次排序merge阶段所有数据处理完毕之后MapTask会对所有临时文件进行一次合并确保只生成一个数据文件
Spill阶段详情
通过快速排序对环形缓冲区内的数据进行排序先按照partition(后面会介绍)编号进行排序然后再按照K进行排序。排序过后数据以分区为单位聚集分区内的所有数按照K有序按照分区编号由小到大将分区数据写入工作目录下的临时文件 output/spillN.out(N表示当前溢写的次数)如果设置了combiner(后面会介绍)则写入文件之前还会将分区中的数据进行一次聚集操作将分区数据的元数据写入到内存索引数据结构SpillRecord中每个分区的元数据包括临时文件的偏移量、压缩前后的数据大小如果内存索引大于1MB会将内存索引写到文件 output.spillN.out.index中
3.3.2 Partition分区
在MapReduce计算模型中Map任务会将生成的键值对按照键进行排序并将其划分到不同的分区中。分区的数量通常等于Reduce任务的数量。具体来说Map任务会按照Partitioner函数定义的分区规则对键值对进行划分。Partitioner函数将每个键值对映射到一个分区编号然后Map任务将其输出到对应的分区中。
Partitioner函数通常是由用户自定义实现的其作用是将键值对映射到一个特定的分区。Hadoop提供了默认的Partitioner实现即HashPartitioner它将键哈希后取模得到分区编号从而实现对键值对的划分。在实际应用中用户可以根据自己的需求自定义Partitioner函数以便将键值对划分到特定的分区中。
如果ReduceTask数量 Partition数量会产生多个空的输出文件
如果ReduceTask数量 Partition数量会导致有分区的数据无处安放会Exception
如果ReduceTask数量 1则不管有多少个分区文件最终都会只产生一个文件
3.3.3 Combiner合并
Combiner在每个mapTask所在的节点运行
Combiner对每个MapTask的输出进行局部汇总减少reduce阶段的负担
Combiner使用的前提是不能影响业务逻辑
3.3.4 ReduceTask工作机制 Copy拉取数据Reduce进程启动copy进程(Fetcher)通过HTTP的方式请求maptask获取自己的文件map task分区表示每个map task属于哪个reduce task MergeReduceTask启动两个线程对内存和磁盘中的文件进行合并防止文件过多。当内存中的数据达到一定阈值时就会启动内存到磁盘的merge与map端的相似。直到没有map端的数据才结束 合并排序将数据合并成一个大数据并进行排序 对排序后的数据调用reduce方法对键相同的键值对调用reduce方法每次调用会产生零个或多个键值对最后将输出的键值对存入HDFS。
3.3.5 shuffle机制 Shuffle阶段的过程可以分为三个阶段
Map端的输出Map任务将生成的键值对按照键排序并将其划分到不同的分区中。如果Map任务的输出缓存区已满则需要将其溢出到本地磁盘的临时文件中。 数据传输在Shuffle阶段中Map任务的输出需要传输到Reduce任务所在的节点以便Reduce任务可以从中提取和合并数据。数据传输是Shuffle阶段的关键步骤其速度和效率直接影响整个MapReduce作业的性能。 Reduce端的输入Reduce任务需要从本地磁盘读取属于自己的分区的临时文件并对同一个分区中的键值对进行合并和排序。Reduce任务将合并后的结果输出到最终的输出文件中。 Shuffle阶段是MapReduce计算模型中非常重要的一个阶段它的性能和效率对整个作业的执行时间和性能影响非常大。因此优化Shuffle阶段的性能和效率是MapReduce应用程序优化的一个关键方向。
3.3.6 排序的解释
MapTask和ReduceTask对key进行排序是为了方便后续的数据处理和计算。
具体来说对于MapTask而言对输出的key进行排序可以将具有相同key值的记录聚合在一起方便ReduceTask进行处理。
而对于ReduceTask而言对输入的key进行排序可以让具有相同key值的记录相邻排列方便进行聚合和计算。
一般来说在Map任务中对键值对进行快速排序的次数是一次即将数据写入环形缓冲区之前对其中的键值对进行排序。这是因为对于同一个Map任务的输出在Map输出的环形缓冲区中进行快速排序即可满足Reduce任务在Shuffle阶段的需求而不需要进行额外的排序。
在Shuffle阶段如果存在多个环形缓冲区需要合并Reduce任务会对它们进行归并排序。这是因为不同Map任务的输出在Shuffle阶段需要合并而这些输出之间的顺序是无序的因此需要进行排序以便进行合并。这次排序是对整个数据集进行的而不是对单个Map任务的输出进行的。
当Reduce任务接收到来自多个Map任务的中间结果时它会对同一个分区内的所有键值对进行排序。这里采用的排序算法一般也是归并排序因为归并排序的时间复杂度是O(nlogn)且适合对大量数据进行排序。这次排序也是对整个数据集进行的而不是对单个Map任务的输出进行的。
因此总体来说Shuffle阶段需要进行多次排序具体排序的次数可能因具体实现而有所不同。但无论是哪种具体实现Shuffle阶段都需要对整个数据集进行排序以便后续的计算和处理。
4 数据压缩
4.1 压缩算法对比 4.2 压缩位置选择
