大数据技术之_05_Hadoop学习_03_MapReduce_MapTask工作机制+ReduceTask工作机制+OutputFormat数据输出+Join多种应用+计数器应用+数据清洗(ETL)

本文主要是介绍大数据技术之_05_Hadoop学习_03_MapReduce_MapTask工作机制+ReduceTask工作机制+OutputFormat数据输出+Join多种应用+计数器应用+数据清洗(ETL),希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

大数据技术之_05_Hadoop学习_03_MapReduce

        • 3.3.4 WritableComparable排序
        • 3.3.5 WritableComparable排序案例实操(全排序)
        • 3.3.6 WritableComparable排序案例实操(区内排序)
        • 3.3.7 Combiner合并
        • 3.3.8 Combiner合并案例实操
        • 3.3.9 GroupingComparator分组(辅助排序/分组排序)
        • 3.3.10 GroupingComparator分组案例实操
      • 3.4 MapTask工作机制
      • 3.5 ReduceTask工作机制
      • 3.6 OutputFormat数据输出
        • 3.6.1 OutputFormat接口实现类
        • 3.6.2 自定义OutputFormat
        • 3.6.3 自定义OutputFormat案例实操
      • 3.7 Join多种应用
        • 3.7.1 Reduce Join
        • 3.7.2 Reduce Join案例实操
        • 3.7.3 Map Join
        • 3.7.4 Map Join案例实操
      • 3.8 计数器应用
      • 3.9 数据清洗(ETL)
        • 3.9.1 数据清洗案例实操-简单解析版
        • 3.9.2 数据清洗案例实操-复杂解析版
        • 3.10 MapReduce开发总结
    • 第4章 Hadoop数据压缩
      • 4.1 概述
      • 4.2 MR支持的压缩编码
      • 4.3 压缩方式选择
        • 4.3.1 Gzip压缩
        • 4.3.2 Bzip2压缩
        • 4.3.3 Lzo压缩
        • 4.3.4 Snappy压缩
      • 4.4 压缩位置选择
      • 4.5 压缩参数配置
      • 4.6 压缩实操案例
        • 4.6.1 数据流的压缩和解压缩
        • 4.6.2 Map输出端采用压缩
        • 4.6.3 Reduce输出端采用压缩
    • 第5章 Yarn资源调度器
      • 5.1 Yarn基本架构
      • 5.3 Yarn工作机制
      • 5.4 作业提交全过程
      • 5.5 资源调度器
      • 5.6 任务的推测执行(秘籍)

3.3.4 WritableComparable排序

0、排序概述


1、排序的分类

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

@Override
public int compareTo(FlowBean o) {int result;// 按照总流量大小,倒序排列if (sumFlow > bean.getSumFlow()) {result = -1;} else if (sumFlow < bean.getSumFlow()) {result = 1;} else {result = 0;}return result;
}
3.3.5 WritableComparable排序案例实操(全排序)

1、需求
  根据案例2.3产生的结果再次对总流量进行排序。
(1)输入数据

(2)期望输出数据

13509468723     7335	110349	117684
13736230513     2481	24681	27162
13956435636     132		1512	1644
13846544121     264		0		264
......

2、需求分析

3、代码实现
(1)FlowBean对象在在需求1基础上增加了比较功能

package com.atguigu.mr.sort;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;import org.apache.hadoop.io.WritableComparable;public class FlowBean implements WritableComparable<FlowBean> {private long upFlow; // 上行流量private long downFlow; // 下行流量private long sumFlow; // 总流量/*** 反序列化时,需要反射调用空参构造函数,所以必须有*/public FlowBean() {super();}public FlowBean(long upFlow, long downFlow) {super();this.upFlow = upFlow;this.downFlow = downFlow;this.sumFlow = upFlow + downFlow;}/*** 序列化方法*/@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(downFlow);out.writeLong(sumFlow);}/*** 反序列化方法,注意反序列化的顺序和序列化的顺序完全一致*/@Overridepublic void readFields(DataInput in) throws IOException {upFlow = in.readLong();downFlow = in.readLong();sumFlow = in.readLong();}/*** 比较方法*/@Overridepublic int compareTo(FlowBean bean) {int result;// 按照总流量大小,倒序排列if (sumFlow > bean.getSumFlow()) {result = -1;} else if (sumFlow < bean.getSumFlow()) {result = 1;} else {result = 0;}return result;}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(long sumFlow) {this.sumFlow = sumFlow;}@Overridepublic String toString() {return upFlow + "\t" + downFlow + "\t" + sumFlow;}
}

(2)编写Mapper类

package com.atguigu.mr.sort;import java.io.IOException;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;public class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text> {FlowBean k = new FlowBean();Text v = new Text();@Overrideprotected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {// 13736230513	2481	24681	27162// 1、获取一行String line = value.toString();// 2、截取String[] fields = line.split("\t");// 3、封装对象String phoneNum = fields[0];long upFlow = Long.parseLong(fields[1]);long downFlow = Long.parseLong(fields[2]);long sumFlow = Long.parseLong(fields[3]);k.setUpFlow(upFlow);k.setDownFlow(downFlow);k.setSumFlow(sumFlow);v.set(phoneNum);// 4、输出context.write(k, v);}
}

(3)编写Reducer类

package com.atguigu.mr.sort;import java.io.IOException;import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;public class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean> {@Overrideprotected void reduce(FlowBean key, Iterable<Text> values, Context context)throws IOException, InterruptedException {// 循环输出,避免总流量相同的情况for (Text text : values) {context.write(text, key);}}
}

(4)编写Driver类

package com.atguigu.mr.sort;import java.io.IOException;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;public class FlowCountSortDriver {public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {// 输入输出路径需要根据自己电脑上实际的输入输出路径设置args = new String[] { "d:/temp/atguigu/0529/output2", "d:/temp/atguigu/0529/output8" };// 1、获取配置信息,或者job对象实例Configuration configuration = new Configuration();Job job = Job.getInstance(configuration);// 2、指定本程序的jar包所在的本地路径job.setJarByClass(FlowCountSortDriver.class);// 3、指定本业务job要使用的mapper/Reducer业务类job.setMapperClass(FlowCountSortMapper.class);job.setReducerClass(FlowCountSortReducer.class);// 4、指定mapper输出数据的kv类型job.setMapOutputKeyClass(FlowBean.class);job.setMapOutputValueClass(Text.class);// 5、指定最终输出的数据的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);// 6、指定job的输入原始文件所在目录FileInputFormat.setInputPaths(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));// 7、将job中配置的相关参数,以及job所用的java类所在的jar包, 提交给yarn去运行boolean result = job.waitForCompletion(true);System.exit(result ? 0 : 1);}
}
3.3.6 WritableComparable排序案例实操(区内排序)

1、需求
  要求每个省份手机号输出的文件中按照总流量内部排序。
2、需求分析
  基于前一个需求,增加自定义分区类,分区按照省份手机号设置。

3、案例实操
(1)增加自定义分区类

package com.atguigu.mr.sort;import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;public class ProvincePartitioner extends Partitioner<FlowBean, Text> {@Overridepublic int getPartition(FlowBean key, Text value, int numPartitions) {// 按照手机号的前三位进行分区// 获取手机号的前三位String prePhoneNum = value.toString().substring(0, 3);// 根据手机号归属地设置分区int partition = 4;if ("136".equals(prePhoneNum)) {partition = 0;} else if ("137".equals(prePhoneNum)) {partition = 1;} else if ("138".equals(prePhoneNum)) {partition = 2;} else if ("139".equals(prePhoneNum)) {partition = 3;}return partition;}
}

(2)在驱动类中添加加载分区类

    // 加载自定义分区类(即关联分区)job.setPartitionerClass(ProvincePartitioner.class);// 设置Reducetask个数job.setNumReduceTasks(5);
3.3.7 Combiner合并

  Combiner合并是Hadoop框架优化的一种手段,因为Combiner合并减少了数据的IO传输。

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

package com.atguigu.mr.wordcount;import java.io.IOException;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {int sum;IntWritable v = new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {// 1、汇总,累加求和sum = 0;for (IntWritable value : values) {sum += value.get();}v.set(sum);// 2、写出context.write(key, v);}
}

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

    job.setCombinerClass(WordcountCombiner.class);
3.3.8 Combiner合并案例实操

1、需求
  统计过程中对每一个MapTask的输出进行局部汇总,以减小网络传输量即采用Combiner功能。
(1)数据输入

banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang

(2)期望输出数据
  期望:Combine输入数据多,输出时经过合并,输出数据降低。
2、需求分析

3、案例实操-方案一
1)增加一个WordcountCombiner类继承Reducer

package com.atguigu.mr.wordcount;import java.io.IOException;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {int sum;IntWritable v = new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {// 1、汇总,累加求和sum = 0;for (IntWritable value : values) {sum += value.get();}v.set(sum);// 2、写出context.write(key, v);}
}

2)在WordcountDriver驱动类中指定Combiner

    // 指定需要使用Combiner,以及用哪个类作为Combiner的逻辑job.setCombinerClass(WordcountCombiner.class);

4、案例实操-方案二
1)将WordcountReducer作为Combiner在WordcountDriver驱动类中指定

    // 指定需要使用Combiner,以及用哪个类作为Combiner的逻辑job.setCombinerClass(WordcountReducer.class);

运行程序,如下图所示:

3.3.9 GroupingComparator分组(辅助排序/分组排序)

  对Reduce阶段的数据根据某一个或几个字段进行分组。
分组排序步骤:
(1)自定义类继承WritableComparator
(2)重写compare()方法

@Override
public int compare(WritableComparable a, WritableComparable b) {// 比较的业务逻辑// ......return result;
}

(3)创建一个构造将比较对象的类传给父类

protected OrderGroupingComparator() {super(OrderBean.class, true);
}
3.3.10 GroupingComparator分组案例实操

1、需求
  有如下订单数据

  现在需要求出每一个订单中最贵的商品。
(1)输入数据
GroupingComparator.txt

0000001	Pdt_01	222.8
0000002	Pdt_05	722.4
0000001	Pdt_02	33.8
0000003	Pdt_06	232.8
0000003	Pdt_02	33.8
0000002	Pdt_03	522.8
0000002	Pdt_04	122.4

(2)期望输出数据

1	222.8
2	722.4
3	232.8

2、需求分析
(1)利用“订单id和成交金额”作为key,可以将Map阶段读取到的所有订单数据按照id升序排序,如果id相同再按照金额降序排序,发送到Reduce。
(2)在Reduce端利用GroupingComparator将订单id相同的kv聚合成组,然后取第一个即是该订单中最贵商品,如下图所示。

3、代码实现
(1)定义订单信息OrderBean类

package com.atguigu.mr.order;import java.io.DataInput;

这篇关于大数据技术之_05_Hadoop学习_03_MapReduce_MapTask工作机制+ReduceTask工作机制+OutputFormat数据输出+Join多种应用+计数器应用+数据清洗(ETL)的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1100803

相关文章

Android ClassLoader加载机制详解

《AndroidClassLoader加载机制详解》Android的ClassLoader负责加载.dex文件,基于双亲委派模型,支持热修复和插件化,需注意类冲突、内存泄漏和兼容性问题,本文给大家介... 目录一、ClassLoader概述1.1 类加载的基本概念1.2 android与Java Class

SQL中如何添加数据(常见方法及示例)

《SQL中如何添加数据(常见方法及示例)》SQL全称为StructuredQueryLanguage,是一种用于管理关系数据库的标准编程语言,下面给大家介绍SQL中如何添加数据,感兴趣的朋友一起看看吧... 目录在mysql中,有多种方法可以添加数据。以下是一些常见的方法及其示例。1. 使用INSERT I

Python中将嵌套列表扁平化的多种实现方法

《Python中将嵌套列表扁平化的多种实现方法》在Python编程中,我们常常会遇到需要将嵌套列表(即列表中包含列表)转换为一个一维的扁平列表的需求,本文将给大家介绍了多种实现这一目标的方法,需要的朋... 目录python中将嵌套列表扁平化的方法技术背景实现步骤1. 使用嵌套列表推导式2. 使用itert

Python使用vllm处理多模态数据的预处理技巧

《Python使用vllm处理多模态数据的预处理技巧》本文深入探讨了在Python环境下使用vLLM处理多模态数据的预处理技巧,我们将从基础概念出发,详细讲解文本、图像、音频等多模态数据的预处理方法,... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核

C#读写文本文件的多种方式详解

《C#读写文本文件的多种方式详解》这篇文章主要为大家详细介绍了C#中各种常用的文件读写方式,包括文本文件,二进制文件、CSV文件、JSON文件等,有需要的小伙伴可以参考一下... 目录一、文本文件读写1. 使用 File 类的静态方法2. 使用 StreamReader 和 StreamWriter二、二进

Python使用pip工具实现包自动更新的多种方法

《Python使用pip工具实现包自动更新的多种方法》本文深入探讨了使用Python的pip工具实现包自动更新的各种方法和技术,我们将从基础概念开始,逐步介绍手动更新方法、自动化脚本编写、结合CI/C... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核

在Linux中改变echo输出颜色的实现方法

《在Linux中改变echo输出颜色的实现方法》在Linux系统的命令行环境下,为了使输出信息更加清晰、突出,便于用户快速识别和区分不同类型的信息,常常需要改变echo命令的输出颜色,所以本文给大家介... 目python录在linux中改变echo输出颜色的方法技术背景实现步骤使用ANSI转义码使用tpu

PostgreSQL的扩展dict_int应用案例解析

《PostgreSQL的扩展dict_int应用案例解析》dict_int扩展为PostgreSQL提供了专业的整数文本处理能力,特别适合需要精确处理数字内容的搜索场景,本文给大家介绍PostgreS... 目录PostgreSQL的扩展dict_int一、扩展概述二、核心功能三、安装与启用四、字典配置方法

MySQL 删除数据详解(最新整理)

《MySQL删除数据详解(最新整理)》:本文主要介绍MySQL删除数据的相关知识,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录一、前言二、mysql 中的三种删除方式1.DELETE语句✅ 基本语法: 示例:2.TRUNCATE语句✅ 基本语

Spring事务传播机制最佳实践

《Spring事务传播机制最佳实践》Spring的事务传播机制为我们提供了优雅的解决方案,本文将带您深入理解这一机制,掌握不同场景下的最佳实践,感兴趣的朋友一起看看吧... 目录1. 什么是事务传播行为2. Spring支持的七种事务传播行为2.1 REQUIRED(默认)2.2 SUPPORTS2