Hadoop学习---7、OutputFormat数据输出、MapReduce内核源码解析、Join应用、数据清洗、MapReduce开发总结(二)

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
云解析 DNS,旗舰版 1个月
全局流量管理 GTM,标准版 1个月
简介: Hadoop学习---7、OutputFormat数据输出、MapReduce内核源码解析、Join应用、数据清洗、MapReduce开发总结(二)

3、Join应用

3.1 Reduce Join

(1)Map端的主要工作:为来自不同表或文件的key/value对,打标签以区别不同来源的记录。然后用连接字段作为key,其余部分和新加的标志作为value,最后进行输出。

(2)Reduce端的主要工作:在Reduce端以连接字段作为key的分组已经完成,我们只需要在每一个分组当中将那些来源于不同文件的记录(在Map阶段已经打标志)分开,最后进行合并就ok了。

3.2 Reduce Join实操

1、需求

3f08e09b5ac044ad83cd46b08a93e2a0.png


aca340e5f5c34b0286b62ddd6a125e64.png


78fa9e5763cc4e6482d3548a2a59b7ed.png


d2cfb74aa83949638088eaaae14f337f.png

2、需求分析

通过将关联条件作为Map输出的key,将两表满足Join条件的数据并携带数据所来源的文件信息,发往同一个ReduceTask,在Reduce中进行数据的串联。

image.png

3、代码实现

(1)创建商品和订单合并后的TableBean类

package org.example._09ReduceJoin;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
/**
 * @ClassName TableBean
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/22 15:09
 * @Version 1.0
 */
public class TableBean implements Writable {
    //订单表id
    private int id;
    //公司id
    private int pid;
    //库存数量
    private int amount;
    //公司名字
    private String pname;
    //表名
    private String flag;
    public TableBean() {
    }
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public int getPid() {
        return pid;
    }
    public void setPid(int pid) {
        this.pid = pid;
    }
    public int getAmount() {
        return amount;
    }
    public void setAmount(int amount) {
        this.amount = amount;
    }
    public String getPname() {
        return pname;
    }
    public void setPname(String pname) {
        this.pname = pname;
    }
    public String getFlag() {
        return flag;
    }
    public void setFlag(String flag) {
        this.flag = flag;
    }
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeInt(id);
        dataOutput.writeInt(pid);
        dataOutput.writeInt(amount);
        dataOutput.writeUTF(pname);
        dataOutput.writeUTF(flag);
    }
    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.id=dataInput.readInt();
        this.pid=dataInput.readInt();
        this.amount=dataInput.readInt();
        this.pname= dataInput.readUTF();
        this.flag=dataInput.readUTF();
    }
    @Override
    public String toString() {
        return id+"\t"+pname+"\t"+amount;
    }
}

(2)创建Mapper类

package org.example._09ReduceJoin;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import java.io.IOException;
/**
 * @ClassName TableMapper
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/22 15:13
 * @Version 1.0
 */
public class TableMapper extends Mapper<LongWritable, Text, IntWritable,TableBean> {
    //pid
    private IntWritable outK=new IntWritable();
    private TableBean outV=new TableBean();
    private String filename;
    //每一个切片进一次
    @Override
    protected void setup(Mapper<LongWritable, Text, IntWritable, TableBean>.Context context) throws IOException, InterruptedException {
        //获取对应文件名称
        InputSplit inputSplit = context.getInputSplit();
        FileSplit fileSplit= (FileSplit) inputSplit;
        filename = fileSplit.getPath().getName();
    }
    //每一行进一次
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, IntWritable, TableBean>.Context context) throws IOException, InterruptedException {
        String line = value.toString();
        //判断是哪个文件,然后针对文件进行不同的操作
        if (filename.contains("order")){//订单表的处理
            String[] split = line.split("\t");
            //封装outK
            outK.set(Integer.parseInt(split[1]));
            //封装outV
            outV.setId(Integer.parseInt(split[0]));
            outV.setPid(Integer.parseInt(split[1]));
            outV.setAmount(Integer.parseInt(split[2]));
            outV.setPname("");
            outV.setFlag("order");
        }else {  //pd表的处理
            String[] split = line.split("\t");
            //封装outK
            outK.set(Integer.parseInt(split[0]));
            //封装outV
            outV.setId(0);
            outV.setPid(Integer.parseInt(split[0]));
            outV.setAmount(0);
            outV.setPname(split[1]);
            outV.setFlag("pd");
        }
        context.write(outK,outV);
    }
}

(3)创建Reduce类

package org.example._09ReduceJoin;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.beanutils.BeanUtilsBean;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
/**
 * @ClassName TableReducer
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/22 15:27
 * @Version 1.0
 */
public class TableReducer extends Reducer<IntWritable,TableBean,TableBean, NullWritable> {
    //每次key相同进来一次
    @Override
    protected void reduce(IntWritable key, Iterable<TableBean> values, Reducer<IntWritable, TableBean, TableBean, NullWritable>.Context context) throws IOException, InterruptedException {
        ArrayList<TableBean> orderBeans=new ArrayList<>();
        TableBean pdBean=new TableBean();
        for (TableBean value : values) {
            if ("order".equals(value.getFlag())){//订单表
                //创建一个临时TableBean对象接受Value
                TableBean tmpOrderBean=new TableBean();
                try {
                    //将value对象复制给tmpOrderBean
                    BeanUtils.copyProperties(tmpOrderBean,value);
                } catch (IllegalAccessException e) {
                    throw new RuntimeException(e);
                } catch (InvocationTargetException e) {
                    throw new RuntimeException(e);
                }
                //将临时TableBean对象添加到集合orderBeans
                orderBeans.add(tmpOrderBean);
            }else { //商品表
                try {
                    BeanUtils.copyProperties(pdBean,value);
                } catch (IllegalAccessException e) {
                    throw new RuntimeException(e);
                } catch (InvocationTargetException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        //便利集合orderBeans,替换掉每个orderBean的pid为pname,然后写出
        for (TableBean orderBean : orderBeans) {
            orderBean.setPname(pdBean.getPname());
            //写出修改后的orderBean对象
            context.write(orderBean,NullWritable.get());
        }
    }
}

(4)创建Driver类

package org.example._09ReduceJoin;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
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;
/**
 * @ClassName TableDriver
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/22 15:39
 * @Version 1.0
 */
public class TableDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration configuration=new Configuration();
        Job job=Job.getInstance(configuration);
        job.setJarByClass(TableDriver.class);
        job.setMapperClass(TableMapper.class);
        job.setReducerClass(TableReducer.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(TableBean.class);
        job.setOutputKeyClass(TableBean.class);
        job.setOutputValueClass(NullWritable.class);
        FileInputFormat.addInputPath(job,new Path("E:\\test\\input4"));
        FileOutputFormat.setOutputPath(job,new Path("E:\\test\\output8"));
        System.exit(job.waitForCompletion(true)?0:1);
    }
}

4、测试

运行结果

1004 小米 4
1001 小米 1
1005 华为 5
1002 华为 2
1006 格力 6
1003 格力 3

5、总结

缺点:这种方式中,合并的操作是在Reduce阶段完成的,Reduce端的处理压力太大,Map节点的运算负载很低,资源利用率不高,且在Reduce阶段极易产生数据倾斜。

解决方法:Map端实现数据合并。

3.3 Mapper Join

1、使用场景

Map Join适合于一张表很小、一张表很大的场景。

2、优点:

在 Map 端缓存多张表,提前处理业务逻辑,这样增加 Map 端业务,减少 Reduce 端数据的压力,尽可能的减少数据倾斜。

3、具体办法

(1)在 Mapper 的 setup 阶段,将文件读取到缓存集合中。

(2)在 Driver 驱动类中加载缓存。

//缓存普通文件到 Task 运行节点。
job.addCacheFile(new URI("file:///e:/cache/pd.txt"));
//如果是集群运行,需要设置 HDFS 路径
job.addCacheFile(new URI("hdfs://hadoop102:8020/cache/pd.txt"));

3.4 Map Join案例实操

1、需求


7b0e851872c14334aa2ac18417d070ec.png

2、需求分析

image.png

3、实现代码

(1)先在Driver中添加缓存文件

package org.example._10MapperJoin;
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;
import java.net.URI;
import java.net.URISyntaxException;
/**
 * @ClassName MapJoinDriver
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/22 16:16
 * @Version 1.0
 */
public class MapJoinDriver {
    public static void main(String[] args) throws IOException, URISyntaxException, InterruptedException, ClassNotFoundException {
        // 1 获取 job 信息
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        // 2 设置加载 jar 包路径
        job.setJarByClass(MapJoinDriver.class);
        // 3 关联 mapper
        job.setMapperClass(MapJoinMapper.class);
        // 4 设置 Map 输出 KV 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        // 5 设置最终输出 KV 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        // 加载缓存数据
        job.addCacheFile(new URI("file:///E:/test/input4/pd.txt"));
        // Map 端 Join 的逻辑不需要 Reduce 阶段,设置 reduceTask 数量为 0
        job.setNumReduceTasks(0);
        // 6 设置输入输出路径
        FileInputFormat.setInputPaths(job, new Path("E:\\test\\input5"));
        FileOutputFormat.setOutputPath(job, new Path("E:\\test\\output9"));
        // 7 提交
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

(2)Mapper(在setup方法中读取缓冲文件)

package org.example._10MapperJoin;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
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.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
/**
 * @ClassName MapJoinMapper
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/22 16:17
 * @Version 1.0
 */
public class MapJoinMapper extends Mapper<LongWritable, Text,Text, NullWritable> {
    private Map<String,String> pdMap=new HashMap<>();
    Text text=new Text();
    //任务开始之前将pd数据缓存进pdMap
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //通过缓存文件得到小表数据pd.txt
        URI[] cacheFiles = context.getCacheFiles();
        Path path=new Path(cacheFiles[0]);
        //获取文件系统对象,并开流
        FileSystem fileSystem = FileSystem.get(context.getConfiguration());
        FSDataInputStream open = fileSystem.open(path);
        //通过包装流转换为reader,方便按行读取
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(open, "UTF-8"));
        //逐行读取,按行处理
        String line;
        while (StringUtils.isNotEmpty(line=bufferedReader.readLine())){
            //切割一行
            //01 小米
            String[] split = line.split("\t");
            pdMap.put(split[0],split[1]);
        }
        //关流
        IOUtils.closeStream(bufferedReader);
    }
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        //读取大表数据
        String[] split = value.toString().split("\t");
        //通过大表每行数据的pid,去pdMap里面取出pname
        String name=pdMap.get(split[1]);
        //将大表每行数据的pid替换为pname
        text.set(split[0]+"\t"+name+"\t"+split[2]);
        context.write(text,NullWritable.get());
    }
}

4、数据清洗

“ETL,是英文 Extract-Transform-Load 的缩写,用来描述将数据从来源端经过抽取(Extract)、转换(Transform)、加载(Load)至目的端的过程。ETL 一词较常用在数据仓库,但其对象并不限于数据仓库。

在运行核心业务 MapReduce 程序之前,往往要先对数据进行清洗,清理掉不符合用户要求的数据。清理的过程往往只需要运行 Mapper 程序,不需要运行 Reduce 程序。

1、需求

去除日志中字段个数小于等于 11 的日志。

(1)输入数据

链接:https://pan.baidu.com/s/1z_nM2e3JrHHZL_5WfbxjaQ

提取码:zhm6

2、需求分析

需要在 Map 阶段对输入的数据根据规则进行过滤清洗。

3、实现代码

(1)Mapper类

package org.example._11ETL;
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 org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;
import java.io.IOException;
/**
 * @ClassName ETLMapper
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/22 16:33
 * @Version 1.0
 */
public class ETLMapper extends Mapper<LongWritable, Text,Text, NullWritable> {
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        //1、获取一行数据
        String line=value.toString();
        //2、解析日志
        boolean result=parseLog(line,context);
        //3、日志不合法
        if (!result){
            return;
        }
        //4、日志合法就直接写出
        context.write(value,NullWritable.get());
    }
    //封装解析日志的方法
    private boolean parseLog(String line, Mapper<LongWritable, Text, Text, NullWritable>.Context context) {
        //1、截取
        String[] fields = line.split(" ");
        //2、日志长度大于11的为合法
        if (fields.length>11){
            return true;
        }
        else {
            return false;
        }
    }
}

(2)Driver类

package org.example._11ETL;
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 org.example._08outputformat.LogDriver;
import java.io.IOException;
/**
 * @ClassName ETLDriver
 * @Description TODO
 * @Author Zouhuiming
 * @Date 2023/5/22 16:39
 * @Version 1.0
 */
public class ETLDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
        args = new String[] { "E:\\test\\input6", "E:\\test\\output10" };
// 1 获取 job 信息
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
// 2 加载 jar 包
        job.setJarByClass(LogDriver.class);
// 3 关联 map
        job.setMapperClass(ETLMapper.class);
// 4 设置最终输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
// 设置 reducetask 个数为 0
        job.setNumReduceTasks(0);
// 5 设置输入和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 6 提交
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

5、MapReduce开发总结

1、输入数据接口:InputFormat

(1)默认使用的实现类是TextInputFormat

(2)TextInputFormat的功能逻辑是:一次读一行文本,然后将该行的起始偏移量作为key,行内容作为value返回。

(3)CombineTextInputFormat可以把多个小文件合并成一个切片处理,提高处理效率。

2、逻辑处理接口:Mapper

用户根据业务需求实现其中的三个方法:map() setup() cleanup()

3、Partitioner分区

(1)有默认实现HashPartitioner,逻辑是根据key的哈希值和numReduces来返回一个分区号;key.hashCode()&Integer.MAXVALUE % numReduces

(2)如果业务上有特别的需求,可以自定义分区。

4、Comparable排序

(1)当我们用自定义的对象作为key来输出时,就必须要实现WritableComparable接口,重写其中的compareTo()方法。

(2)部分排序:对最输出的每一个文件内部进行排序。

(3)全排序:对所以数据进行排序,通常只有一个Reduce。

(4)二次排序:排序的条件有两个。

5、Combiner合并

Combiner合并可以提高程序执行效率,减少IO传输。但是使用时必须不能影响原有的业务处理结果。

6、逻辑处理接口:Reducer

用户根据业务需求实现其中的三个方法: reduce() setup() cleanup()

7、输出数据接口:OutputFormat

(1)默认实现类是TextOutputFormat,功能逻辑是:将每一个KV对,向目标文本文件输出一行。

(2)用户还可以自定义OutputFormat。


相关文章
|
2月前
|
缓存 分布式计算 算法
优化Hadoop MapReduce性能的最佳实践
【8月更文第28天】Hadoop MapReduce是一个用于处理大规模数据集的软件框架,适用于分布式计算环境。虽然MapReduce框架本身具有很好的可扩展性和容错性,但在某些情况下,任务执行可能会因为各种原因导致性能瓶颈。本文将探讨如何通过调整配置参数和优化算法逻辑来提高MapReduce任务的效率。
186 0
|
4月前
|
分布式计算 Hadoop Java
Hadoop MapReduce编程
该教程指导编写Hadoop MapReduce程序处理天气数据。任务包括计算每个城市ID的最高、最低气温、气温出现次数和平均气温。在读取数据时需忽略表头,且数据应为整数。教程中提供了环境变量设置、Java编译、jar包创建及MapReduce执行的步骤说明,但假设读者已具备基础操作技能。此外,还提到一个扩展练习,通过分区功能将具有相同尾数的数字分组到不同文件。
47 1
|
4月前
|
数据采集 SQL 分布式计算
|
4月前
|
存储 分布式计算 Hadoop
Hadoop生态系统详解:HDFS与MapReduce编程
Apache Hadoop是大数据处理的关键,其核心包括HDFS(分布式文件系统)和MapReduce(并行计算框架)。HDFS为大数据存储提供高容错性和高吞吐量,采用主从结构,通过数据复制保证可靠性。MapReduce将任务分解为Map和Reduce阶段,适合大规模数据集的处理。通过代码示例展示了如何使用MapReduce实现Word Count功能。HDFS和MapReduce的结合,加上YARN的资源管理,构成处理和分析大数据的强大力量。了解和掌握这些基础对于有效管理大数据至关重要。【6月更文挑战第12天】
125 0
|
4月前
|
分布式计算 Java Hadoop
简单的java Hadoop MapReduce程序(计算平均成绩)从打包到提交及运行
简单的java Hadoop MapReduce程序(计算平均成绩)从打包到提交及运行
43 0
|
4月前
|
分布式计算 Hadoop Java
使用Hadoop MapReduce分析邮件日志提取 id、状态 和 目标邮箱
使用Hadoop MapReduce分析邮件日志提取 id、状态 和 目标邮箱
|
2月前
|
监控 网络协议 Java
Tomcat源码解析】整体架构组成及核心组件
Tomcat,原名Catalina,是一款优雅轻盈的Web服务器,自4.x版本起扩展了JSP、EL等功能,超越了单纯的Servlet容器范畴。Servlet是Sun公司为Java编程Web应用制定的规范,Tomcat作为Servlet容器,负责构建Request与Response对象,并执行业务逻辑。
Tomcat源码解析】整体架构组成及核心组件
|
2月前
|
存储 NoSQL Redis
redis 6源码解析之 object
redis 6源码解析之 object
58 6
|
21天前
|
存储 缓存 Java
什么是线程池?从底层源码入手,深度解析线程池的工作原理
本文从底层源码入手,深度解析ThreadPoolExecutor底层源码,包括其核心字段、内部类和重要方法,另外对Executors工具类下的四种自带线程池源码进行解释。 阅读本文后,可以对线程池的工作原理、七大参数、生命周期、拒绝策略等内容拥有更深入的认识。
什么是线程池?从底层源码入手,深度解析线程池的工作原理
|
25天前
|
开发工具
Flutter-AnimatedWidget组件源码解析
Flutter-AnimatedWidget组件源码解析

热门文章

最新文章

相关实验场景

更多
下一篇
无影云桌面