shuffle是从第七步开始的到第十六步结束的,下面是shuffle过程详解
Map方法之后,Reduce方法之前的数据处理过程称之为Shuffle。
Partitioner
,重写getPartition()
方法java
package cn.buildworld.mapreduce.partition;
import cn.buildworld.mapreduce.flow.FlowBean;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
/**
* @author MiChong
* @date 2020-05-27 14:40
*/
public class MyPartitioner extends Partitioner<Text, FlowBean> {
@Override
public int getPartition(Text text, FlowBean flowBean, int numPartitions) {
String phone = text.toString();
switch (phone.substring(0, 3)) {
case "136":
return 0;
case "137":
return 1;
case "138":
return 2;
case "139":
return 3;
default:
return 4;
}
}
}
Partitioner
java
job.setPartitionerClass(MyPartitioner.class);
Partition
后,要根据自定义Partitioner
的逻辑设置相应数量的ReduceTask
java
job.setNumReduceTasks(5);
ReduceTask
的数量> getPartition
的结果数,则会多产生几个空的输出文件part-r-000xx;ReduceTask
的数量<getPartition
的结果数,则有一部分分区数据无处安放,会Exception;ReduceTask
的数量=1,则不管MapTask端输出多少个分区文件,最终结果都交给这一个ReduceTask
,最终也就只会产生一个结果文件 part-r-00000;Bean对象实现WritableComparable
几口,重写compareTo()
方法
java
package cn.buildworld.mapreduce.writablecomparable;
import org.apache.hadoop.io.WritableComparable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
/**
* @author MiChong
* @date 2020-05-24 18:15
*/
/**
* 必须实现Writable接口
*/
public class FlowBean implements WritableComparable<FlowBean> {
private long upFlow;
private long downFlow;
private long sumFlow;
// 此处实现自定义排序
@Override
public int compareTo(FlowBean o) {
return Long.compare(o.sumFlow, this.sumFlow);
}
}
自定义一个Combiner继承Reducer,重写Reduce方法
java
public class WordcountCombiner extends Reducer<Text, IntWritable, Text,IntWritable>{
@Override
protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
// 1 汇总操作
int count = 0;
for(IntWritable v :values){
count += v.get();
}
// 2 写出
context.write(key, new IntWritable(count));
}
在Job驱动类中设置
java
job.setCombinerClass(WordcountCombiner.class);
WritableComparator
compare()
方法java
package cn.buildworld.mapreduce.groupCompa;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
/**
* @author MiChong
* @date 2020-05-30 13:44
*/
public class OrderComparator extends WritableComparator {
protected OrderComparator() {
super(OrderBean.class, true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
OrderBean oa = (OrderBean) a;
OrderBean ob = (OrderBean) b;
return oa.getOrderId().compareTo(ob.getOrderId());
}
}
溢写阶段详情:
步骤1:利用快速排序算法对缓存区内的数据进行排序,排序方式是,先按照分区编号Partition进行排序,然后按照key进行排序。这样,经过排序后,数据以分区为单位聚集在一起,且同一分区内所有数据按照key有序。
步骤2:按照分区编号由小到大依次将每个分区中的数据写入任务工作目录下的临时文件output/spillN.out(N表示当前溢写次数)中。如果用户设置了Combiner,则写入文件之前,对每个分区中的数据进行一次聚集操作。
步骤3:将分区数据的元信息写到内存索引数据结构SpillRecord中,其中每个分区的元信息包括在临时文件中的偏移量、压缩前数据大小和压缩后数据大小。如果当前内存索引大小超过1MB,则将内存索引写到文件output/spillN.out.index中。
当所有数据处理完后,MapTask会将所有临时文件合并成一个大文件,并保存到文件output/file.out中,同时生成相应的索引文件output/file.out.index。
在进行文件合并过程中,MapTask以分区为单位进行合并。对于某个分区,它将采用多轮递归合并的方式。每轮合并io.sort.factor(默认10)个文件,并将产生的文件重新加入待合并列表中,对文件排序后,重复以上过程,直到最终得到一个大文件。
让每个MapTask最终只生成一个数据文件,可避免同时打开大量文件和同时读取大量小文件产生的随机读取带来的开销。
educeTask的并行度同样影响整个Job的执行并发度和执行效率,但与MapTask的并发数由切片数决定不同,ReduceTask数量的决定是可以直接手动设置:
java
// 默认值是1,手动设置为4
job.setNumReduceTasks(4);
TextOutputFormat
默认的输出格式是TextOutputFormat,它把每条记录写为文本行。它的键和值可以是任意类型,因为TextOutputFormat调用toString()方法把它们转换为字符串。
SequenceFileOutputFormat
将SequenceFileOutputFormat输出作为后续 MapReduce任务的输入,这便是一种好的输出格式,因为它的格式紧凑,很容易被压缩。
OutputFormat
根据用户需求,自定义实现输出。
java
package cn.buildworld.mapreduce.outputformat;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
/**
* @author MiChong
* @date 2020-05-31 16:45
*/
public class MyRecordWriter extends RecordWriter<LongWritable, Text> {
private FSDataOutputStream atguigu;
private FSDataOutputStream other;
/**
* 初始化方法
*/
public void initialize(TaskAttemptContext job) throws IOException {
String dir = job.getConfiguration().get(FileOutputFormat.OUTDIR);
FileSystem fileSystem = FileSystem.get(job.getConfiguration());
atguigu = fileSystem.create(new Path(dir+"/my.log"));
other = fileSystem.create(new Path(dir+"/others.log"));
}
/**
* 将KV写出,每对KV调用一次
*
* @param key
* @param value
* @throws IOException
* @throws InterruptedException
*/
@Override
public void write(LongWritable key, Text value) throws IOException, InterruptedException {
String out = value.toString() + "\n";
if (out.contains("atguigu")) {
atguigu.write(out.getBytes());
} else {
other.write(out.getBytes());
}
}
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException {
IOUtils.closeStream(atguigu);
IOUtils.closeStream(other);
}
}
Map Join适用于一张表十分小、一张表很大的场景。
思考:在Reduce端处理过多的表,非常容易产生数据倾斜。怎么办?
在Map端缓存多张表,提前处理业务逻辑,这样增加Map端业务,减少Reduce端数据的压力,尽可能的减少数据倾斜。
(1)在Mapper的setup阶段,将文件读取到缓存集合中。
(2)在驱动函数中加载缓存。
java
import java.net.URI;
// 缓存普通文件到Task运行节点。
job.addCacheFile(new URI("file:///e:/cache/pd.txt"));
TextInputFormat
TextInputFormat
的功能逻辑是:一次读一行文本,然后将该行的起始偏移量作为key,行内容作为value返回。KeyValueTextInputFormat
每一行均为一条记录,被分隔符分割为key,value。默认分隔符是tab(\t)。NlineInputFormat
按照指定的行数N来划分切片。CombineTextInputFormat
可以把多个小文件合并成一个切片处理,提高处理效率。InputFormat
。用户根据业务需求实现其中三个方法:map() setup() cleanup ()
Combiner
合并可以提高程序执行效率,减少IO传输。但是使用时必须不能影响原有的业务处理结果。
在Reduce
端对key进行分组。应用于:在接收的key为bean对象时,想让一个或几个字段相同(全部字段比较不相同)的key进入到同一个reduce方法时,可以采用分组排序。
用户根据业务需求实现其中三个方法:reduce() setup() cleanup ()