Hadoop学习:深入解析MapReduce的大数据魔力(二)

IT知识
103
0
0
2024-05-25
Hadoop学习:深入解析MapReduce的大数据魔力(二)

3.3 Shuffle 机制

3.3.1 Shuffle 机制

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

在这里插入图片描述

3.3.2 Partition 分区

1、问题引出 要求将统计结果按照条件输出到不同文件中(分区)。比如:将统计结果按照手机 归属地不同省份输出到不同文件中(分区) 2、默认Partitioner分区

public class HashPartitioner<K, V> extends Partitioner<K, V> {
 }
 public int getPartition(K key, V value, int numReduceTasks) {
 return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
 }

默认分区是根据key的hashCode对ReduceTasks个数取模得到的。用户没法控制哪个key存储到哪个分区。 3、自定义Partitioner步骤 (1)自定义类继承Partitioner,重写getPartition()方法

public class CustomPartitioner extends Partitioner<Text, FlowBean> {
 @Override
 public int getPartition(Text key, FlowBean value, int numPartitions) {
 // 控制分区代码逻辑
… …
 }
 }
 return partition;

(2)在Job驱动中,设置自定义Partitioner

 job.setPartitionerClass(CustomPartitioner.class);

(3)自定义Partition后,要根据自定义Partitioner的逻辑设置相应数量的ReduceTask

 job.setNumReduceTasks(5);

4、分区总结 (1)如果ReduceTask的数量>getPartition的结果数,则会多产生几个空的输出文件part-r-000xx; (2)如果1<ReduceTask的数量<getPartition的结果数,则有一部分分区数据无处安放,会Exception; (3)如果ReduceTask的数量=1,则不管MapTask端输出多少个分区文件,最终结果都交给这一个 ReduceTask,最终也就只会产生一个结果文件part-r-00000; (4)分区号必须从零开始,逐一累加。 5、案例分析 例如:假设自定义分区数为5,则 (1)job.setNumReduceTasks(1); 会正常运行,只不过会产生一个输出文件 (2)job.setNumReduceTasks(2);会报错 (3)job.setNumReduceTasks(6); 大于5,程序会正常运行,会产生空文件

3.3.3 Partition 分区案例实操

1)需求 将统计结果按照手机归属地不同省份输出到不同文件中(分区) (1)输入数据

在这里插入图片描述

2)期望输出数据 手机号136、137、138、139开头都分别放到一个独立的4个文件中,其他开头的放到 一个文件中。 2)需求分析 1、需求:将统计结果按照手机归属地不同省份输出到不同文件中(分区)

在这里插入图片描述

3)在案例2.3的基础上,增加一个分区类

package com.atguigu.mapreduce.partitioner; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Partitioner; 
 
public class ProvincePartitioner extends Partitioner<Text, FlowBean> { 
 
    @Override 
    public int getPartition(Text text, FlowBean flowBean, int numPartitions) 
{ 
        //获取手机号前三位prePhone 
        String phone = text.toString(); 
        String prePhone = phone.substring(0, 3); 
 
        //定义一个分区号变量partition,根据prePhone设置分区号 
        int partition; 
 
        if("136".equals(prePhone)){ 
            partition = 0; 
        }else if("137".equals(prePhone)){ 
            partition = 1; 
        }else if("138".equals(prePhone)){ 
            partition = 2; 
        }else if("139".equals(prePhone)){ 
            partition = 3; 
        }else { 
            partition = 4; 
        } 
 
        //最后返回分区号partition 
        return partition; 
    } 
} 

4)在驱动函数中增加自定义数据分区设置和ReduceTask设置

package com.atguigu.mapreduce.partitioner; 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
import java.io.IOException; 
 
public class FlowDriver { 
 
    public static void main(String[] args) throws IOException, 
ClassNotFoundException, InterruptedException { 
 
        //1 获取job对象 
        Configuration conf = new Configuration(); 
        Job job = Job.getInstance(conf); 
 
        //2 关联本Driver类 
        job.setJarByClass(FlowDriver.class); 
 
        //3 关联Mapper和Reducer 
        job.setMapperClass(FlowMapper.class); 
        job.setReducerClass(FlowReducer.class); 
 
        //4 设置Map端输出数据的KV类型 
        job.setMapOutputKeyClass(Text.class); 
        job.setMapOutputValueClass(FlowBean.class); 
 
        //5 设置程序最终输出的KV类型 
        job.setOutputKeyClass(Text.class); 
        job.setOutputValueClass(FlowBean.class); 
 
        //8 指定自定义分区器 
        job.setPartitionerClass(ProvincePartitioner.class); 
 
        //9 同时指定相应数量的ReduceTask 
        job.setNumReduceTasks(5); 
 
        //6 设置输入输出路径 
        FileInputFormat.setInputPaths(job, new Path("D:\\inputflow")); 
        FileOutputFormat.setOutputPath(job, new Path("D\\partitionout")); 
 
        //7 提交Job 
        boolean b = job.waitForCompletion(true); 
        System.exit(b ? 0 : 1); 
    } 
} 
3.3.4 WritableComparable 排序

排序概述 排序是MapReduce框架中最重要的操作之一。

MapTask和ReduceTask均会对数据按照key进行排序。该操作属于 Hadoop的默认行为。任何应用程序中的数据均会被排序,而不管逻辑上是 否需要。

默认排序是按照字典顺序排序,且实现该排序的方法是快速排序。

对于MapTask,它会将处理的结果暂时放到环形缓冲区中,当环形缓冲区使 用率达到一定阈值后,再对缓冲区中的数据进行一次快速排序,并将这些有序数 据溢写到磁盘上,而当数据处理完毕后,它会对磁盘上所有文件进行归并排序。 对于ReduceTask,它从每个MapTask上远程拷贝相应的数据文件,如果文件大 小超过一定阈值,则溢写磁盘上,否则存储在内存中。如果磁盘上文件数目达到 一定阈值,则进行一次归并排序以生成一个更大文件;如果内存中文件大小或者 数目超过一定阈值,则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完 毕后,ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。

排序分类

(1)部分排序 MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序。 (2)全排序 最终输出结果只有一个文件,且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在 处理大型文件时效率极低,因为一台机器处理所有文件,完全丧失了MapReduce所提供的并行架构。 (3)辅助排序:(GroupingComparator分组) (4)二次排序 在Reduce端对key进行分组。应用于:在接收的key为bean对象时,想让一个或几个字段相同(全部 字段比较不相同)的key进入到同一个reduce方法时,可以采用分组排序。 在自定义排序过程中,如果compareTo中的判断条件为两个即为二次排序。

自定义排序WritableComparable原理分析 bean 对象做为key传输,需要实现WritableComparable接口重写compareTo方法,就可以实现排序。

@Override 
public int compareTo(FlowBean bean) { 
int result; 
// 按照总流量大小,倒序排列 
if (this.sumFlow > bean.getSumFlow()) { 
result = -1; 
}else if (this.sumFlow < bean.getSumFlow()) { 
result = 1; 
}else { 
result = 0; 
} 
return result; 
} 
3.3.5 Combiner 合并

(1)Combiner是MR程序中Mapper和Reducer之外的一种组件。 (2)Combiner组件的父类就是Reducer。 (3)Combiner和Reducer的区别在于运行的位置 Combiner是在每一个MapTask所在的节点运行; Reducer是接收全局所有Mapper的输出结果; 应该跟Reducer的输入kv类型要对应起来。 Mapper (4)Combiner的意义就是对每一个MapTask的输出进行局部汇总,以减小网络传输量。 (5)Combiner能够应用的前提是不能影响最终的业务逻辑,而且,Combiner的输出kv应该跟Reducer的输入kv类型要对应起来。

在这里插入图片描述

(6)自定义Combiner实现步骤 (a)自定义一个Combiner继承Reducer,重写Reduce方法

public class WordCountCombiner extends Reducer<Text, IntWritable, Text, 
IntWritable> { 
private IntWritable outV = new IntWritable(); 
@Override 
protected void reduce(Text key, Iterable<IntWritable> values, Context 
context) throws IOException, InterruptedException { 
int sum = 0; 
for (IntWritable value : values) { 
sum += value.get(); 
} 
outV.set(sum); 
context.write(key,outV); 
} 
} 

(b)在Job驱动类中设置:

job.setCombinerClass(WordCountCombiner.class); 

3.4 OutputFormat 数据输出

3.4.1 OutputFormat 接口实现类

OutputFormat是MapReduce输出的基类,所有实现MapReduce输出都实现了OutputFormat 接口。下面我们介绍几种常见的OutputFormat实现类。 1.OutputFormat实现类 2.默认输出格式TextOutputFormat 3.自定义OutputFormat 3.1 应用场景: 例如:输出数据到MySQL/HBase/Elasticsearch等存储框架中。 3.2 自定义OutputFormat步骤 ➢ 自定义一个类继承FileOutputFormat。 ➢ 改写RecordWriter,具体改写输出数据的方法write()。

3.4.2 自定义OutputFormat案例实操

1)需求 过滤输入的log日志,包含atguigu的网站输出到e:/atguigu.log,不包含atguigu的网站输出到e:/other.log。 (1)输入数据

在这里插入图片描述

(2)期望输出数据

在这里插入图片描述

2)需求分析

在这里插入图片描述

3)案例实操 (1)编写LogMapper类

package com.atguigu.mapreduce.outputformat; 
import org.apache.hadoop.io.LongWritable; 
import org.apache.hadoop.io.NullWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Mapper; 
import java.io.IOException; 
public 
class 
LogMapper 
extends 
Mapper<LongWritable, 
Text,Text, 
NullWritable> { 
@Override 
protected void map(LongWritable key, Text value, Context context) 
throws IOException, InterruptedException { 
//不做任何处理,直接写出一行log数据 
context.write(value,NullWritable.get()); 
} 
} 

(2)编写LogReducer类

package com.atguigu.mapreduce.outputformat; 
 
import org.apache.hadoop.io.NullWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Reducer; 
 
import java.io.IOException; 
 
public class LogReducer extends Reducer<Text, NullWritable,Text, 
NullWritable> { 
    @Override 
    protected void reduce(Text key, Iterable<NullWritable> values, Context 
context) throws IOException, InterruptedException { 
        // 防止有相同的数据,迭代写出 
        for (NullWritable value : values) { 
            context.write(key,NullWritable.get()); 
        } 
    } 
} 

(3)自定义一个LogOutputFormat类

package com.atguigu.mapreduce.outputformat; 
 
import org.apache.hadoop.io.NullWritable; 
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; 
 
public class LogOutputFormat extends FileOutputFormat<Text, NullWritable> 
{ 
    @Override 
    public RecordWriter<Text, NullWritable> 
getRecordWriter(TaskAttemptContext job) throws IOException, 
InterruptedException { 
        //创建一个自定义的RecordWriter返回 
        LogRecordWriter logRecordWriter = new LogRecordWriter(job); 
        return logRecordWriter; 
    } 
} 

(4)编写LogRecordWriter类

package com.atguigu.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.NullWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.RecordWriter; 
import org.apache.hadoop.mapreduce.TaskAttemptContext; 
 
import java.io.IOException; 
 
public class LogRecordWriter extends RecordWriter<Text, NullWritable> { 
 
    private FSDataOutputStream atguiguOut;
      private FSDataOutputStream otherOut; 
 
    public LogRecordWriter(TaskAttemptContext job) { 
        try { 
            //获取文件系统对象 
            FileSystem fs = FileSystem.get(job.getConfiguration()); 
            //用文件系统对象创建两个输出流对应不同的目录 
            atguiguOut = fs.create(new Path("d:/hadoop/atguigu.log")); 
            otherOut = fs.create(new Path("d:/hadoop/other.log")); 
        } catch (IOException e) { 
            e.printStackTrace(); 
        } 
    } 
 
    @Override 
    public void write(Text key, NullWritable value) throws IOException, 
InterruptedException { 
        String log = key.toString(); 
        //根据一行的log数据是否包含atguigu,判断两条输出流输出的内容 
        if (log.contains("atguigu")) { 
            atguiguOut.writeBytes(log + "\n"); 
        } else { 
            otherOut.writeBytes(log + "\n"); 
        } 
    } 
 
    @Override 
    public void close(TaskAttemptContext context) throws IOException, 
InterruptedException { 
        //关流 
        IOUtils.closeStream(atguiguOut); 
        IOUtils.closeStream(otherOut); 
    } 
} 

(5)编写LogDriver类

package com.atguigu.mapreduce.outputformat; 
 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.NullWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
 
import java.io.IOException; 
 
public class LogDriver { 
    public static void main(String[] args) throws IOException, 
ClassNotFoundException, InterruptedException { 
 
    Configuration conf = new Configuration(); 
    Job job = Job.getInstance(conf); 
 
    job.setJarByClass(LogDriver.class); 
    job.setMapperClass(LogMapper.class); 
    job.setReducerClass(LogReducer.class); 
 
    job.setMapOutputKeyClass(Text.class); 
    job.setMapOutputValueClass(NullWritable.class); 
    ——————————————————————————— 
    job.setOutputKeyClass(Text.class); 
    job.setOutputValueClass(NullWritable.class); 
    //设置自定义的outputformat 
    job.setOutputFormatClass(LogOutputFormat.class); 
    FileInputFormat.setInputPaths(job, new Path("D:\\input")); 
    //虽然我们自定义了 outputformat,但是因为我们的 outputformat 继承自
    fileoutputformat 
    //而fileoutputformat 要输出一个_SUCCESS文件,所以在这还得指定一个输出目录 
    FileOutputFormat.setOutputPath(job, new Path("D:\\logoutput")); 
    boolean b = job.waitForCompletion(true); 
    System.exit(b ? 0 : 1); 
  } 
}