MapReduce分布式计算

Posted 2023-05-11 _泡泡

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了MapReduce分布式计算相关的知识，希望对你有一定的参考价值。

MapReduce是Hadoop系统核心组件之一，它是一种可用于大数据并行处理的计算模型、框架和平台，主要解决海量数据的计算，是目前分布式计算模型中应用较为广泛的一种。

练习：计算a.txt文件中每个单词出现的次数

hello world
hello hadoop
hello 51doit
hadoop mapreduce
mapreduce spark

public class WordCount 
    public static void main(String[] args) throws IOException 
        //获取到resource文件夹下a.txt的路径
        URL resource = WordCount.class.getClassLoader().getResource("a.txt");
        String path = resource.getPath();
        //使用FileUtils将文件读取成字符串
        String s = FileUtils.readFileToString(new File(path),"utf-8");
        //将文件使用空格进行切割  \\s可以切割 空格 tab键
        String[] arr = s.split("\\\\s+");

        //创建Map集合
        Map<String,Integer> map = new HashMap<>();

        //遍历数组
        for (String s1 : arr) 
            //判断集合是否包含指定键
            if(!map.containsKey(s1))
                //如果不包含 添加 单词 1
                map.put(s1,1);
            else
                //如果包含  获取当前键的次数 +1 在添加回集合
                Integer count = map.get(s1);
                map.put(s1,count+1);
            
        

        System.out.println(map);

通过以上的方式计算出来了a.txt文件中每个单词出现的次数，但是我们想一下，如果a.txt文件非常大，怎么办？

比如有一个a.txt文件10个T的大小。这时一台计算机就没有办法计算了，因为我们根本存储不了，计算不了，那么一台计算机无法计算，就使用多台计算机来进行计算！

MapReduce核心思想

MapReduce的核心思想是“分而治之”。所谓“分而治之”就是把一个复杂的问题，按照一定的“分解”方法分为等价的规模较小的若干部分，然后逐个解决，分别找出各部分的结果，把各部分的结果组成整个问题的结果，这种思想来源于日常生活与工作时的经验，同样也完全适合技术领域。

为了更好地理解“分而治之”思想，我们光来举一个生活的例子。例如，某大型公司在全国设立了分公司，假设现在要统计公司今年的营收情况制作年报，有两种统计方式，第1种方式是全国分公司将自己的账单数据发送至总部，由总部统一计算公司今年的营收报表:第2种方式是采用分而治之的思想，也就是说，先要求分公司各自统计营收情况，再将统计结果发给总部进行统一汇总计算。这两种方式相比，显然第2种方式的策略更好，工作效率更高效。

MapReduce 作为一种分布式计算模型，它主要用于解决海量数据的计算问题。使用MapReduce操作海量数据时，每个MapReduce程序被初始化为一个工作任务，每个工作任务可以分为Map 和l Reducc两个阶段，具体介绍如下:

Map阶段：:负责将任务分解，即把复杂的任务分解成若干个“简单的任务”来行处理，但前提是这些任务没有必然的依赖关系，可以单独执行任务。

Reduce阶段:负责将任务合并，即把Map阶段的结果进行全局汇总。下面通过一个图来描述上述MapReduce 的核心思想。

MapReduce就是“任务的分解与结和的汇总”。即使用户不懂分布式计算框架的内部运行机制，但是只要能用Map和 Reduce思想描述清楚要处理的问题，就能轻松地在Hadoop集群上实现分布式计算功能。

MapReduce编程模型

MapReduce是一种编程模型，用于处理大规模数据集的并行运算。使用MapReduce执行计算任务的时候，每个任务的执行过程都会被分为两个阶段，分别是Map和Reduce，其中Map阶段用于对原始数据进行处理，Reduce阶段用于对Map阶段的结果进行汇总，得到最终结果。

MapReduce编程模型借鉴了函数式程序设计语言的设计思想，其程序实现过程是通过map()和l reduce()函数来完成的。从数据格式上来看，map()函数接收的数据格式是键值对，生的输出结果也是键值对形式，reduce()函数会将map()函数输出的键值对作为输入，把相同key 值的 value进行汇总，输出新的键值对。

(1）将原始数据处理成键值对<K1，V1>形式。

(2）将解析后的键值对<K1，V1>传给map()函数，map()函数会根据映射规则，将键值对<K1，V1>映射为一系列中间结果形式的键值对<K2，V2>。

(3)将中间形式的键值对<K2，V2>形成<K2，V2，....>形式传给reduce()函数处理，把具有相同key的value合并在一起，产生新的键值对<K3，V3>，此时的键值对<K3，V3>就是最终输出的结果。

词频统计

因为我们的数据都存储在不同的计算机中,那么将对象中的数据从网络中传输,就一定要用到序列化!

/*
	JDK序列化对象的弊端 
	我们进行序列化 其实最主要的目的是为了 序列化对象的属性数据
	比如如果序列化一个Person对象 new Person("柳岩",38); 其实我们想要的是 柳岩 38
	但是如果直接序列化一个对象的话 JDK为了反序列化方便 会在文件中加入其他的数据 这样
	序列化后的文件会变的很大,占用空间
*/
public class Test 
    public static void main(String[] args) throws Exception 
        ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream("d:\\\\person.txt"));

        //JDK序列化对象 
        Person p = new Person();
        p.setName("柳岩");
        p.setAge(38);
        oos.writeObject(p);
        oos.close();

本来其实数据就占几个字节,序列化后,多占用了很多字节,这样如果序列化多的话就会浪费很多空间.

/*
	可以通过序列化属性的方式解决问题
	只序列化属性 可以减小序列化后的文件大小
*/
public class Test 
    public static void main(String[] args) throws Exception 
        ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream("d:\\\\person.txt"));

        Person p = new Person();
        p.setName("柳岩");
        p.setAge(38);

        //只序列化属性
        oos.writeUTF(p.getName());
        oos.writeInt(p.getAge());
        oos.close();

/*
	需要注意
	反序列化时 需要按照序列化的顺序来反序列化
*/
public class Test 
    public static void main(String[] args) throws Exception 
        ObjectInputStream ois = new ObjectInputStream(new FileInputStream("d:\\\\person.txt"));
		//先反序列化name 在反序列化age
        String name = ois.readUTF();
        int age = ois.readInt();
        System.out.println(name + " "+age);
        ois.close();

Hadoop对java的序列化又进行了优化,对一些类型进行了进一步的封装,方便按照自己的方式序列化

Integer  ----> IntWritable
Long     ----> LongWritable
String   ----> Text
Double   ----> DoubleWritable
Boolean  ----> BooleanWritable

WorldCount代码编写

map函数定义

/*
	KEYIN: K1   
	VALUIN: V1  
	KEYOUT:K2   
	VALUEOUT:V2 
*/
public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> 
  
   protected void map(KEYIN key, VALUEIN value, Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>.Context 		context) throws IOException, InterruptedException

我们只需要继承Mapper类,重写map方法就好

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/*
     K1 : 行起始位置   数字 Long  ---- > LongWritable
     V1 : 一行数据  字符串 String  -----> Text
     K2 : 单词     字符串 String  ----->  Text
     V2 : 固定数字1 数组  Long -----> LongWritable
 */
public class WordCountMapper  extends Mapper<LongWritable, Text,Text,LongWritable> 


    /**
     *
     * @param key   K1
     * @param value V1
     * @param context  上下文对象 将map的结果 输出给reduce
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        //将一行数据 转换成字符串 按照空格切割 
        String[] arr = value.toString().split("\\\\s+");
        for (String k2 : arr) 
              //将单词输出给reduce
              context.write(new Text(k2),new LongWritable(1));

reduce函数定义

/*
	KEYIN:K2 
	VALUEIN:V2
	KEYOUT:K3
	VALUEOUT:V3
*/
public class Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> 
  
    protected void reduce(KEYIN key, Iterable<VALUEIN> values, Reducer<KEYIN, VALUEIN, KEYOUT, 					VALUEOUT>.Context context) throws IOException, InterruptedException

我们只需要继承Reducer类型重写reduce方法就好

/*
    K2:单词        String  ----> Text
    V2:固定数字 1   Long    ----> LongWritable
    K3:单词        String  ----> Text
    V3:相加后的结果 Long    ----> LongWritable
 */
public class WordCountReducer extends Reducer<Text,LongWritable,Text,LongWritable> 

    /**
     *
     * @param key  K2
     * @param values  V2的集合 1,1,1,1
     * @param context 上下文对象 输出结果
     */
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException 

        int count = 0;
        //将次数相加
        for (LongWritable value : values) 
               count+=value.get();
        
        //写出 k3 v3
        context.write(key,new LongWritable(count));

最后编写启动程序

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;

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 Test 
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException 
        //创建配置对象
        Configuration conf = new Configuration();
        //创建工作任务
        Job job = Job.getInstance(conf, "wordCount");

        //设置Map类
        job.setMapperClass(WordCountMapper.class);
        //设置Reduce类
        job.setReducerClass(WordCountReducer.class);

        //设置map的输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        //设置reduce的输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        //设置读取文件位置  可以是文件 也可以是文件夹
        FileInputFormat.setInputPaths(job,new Path("d:\\\\work\\\\abc"));
        //设置输出文件位置
        FileOutputFormat.setOutputPath(job,new Path("d:\\\\work\\\\abc\\\\out_put"));

        //提交任务 并等待任务结束
        job.waitForCompletion(true);
    


如果抛这个异常 需要查看windows环境
Exception in thread "main"java.lang .UnsatisfiedLinkError: org.apache .hadoop.io.nativeio.NativeIO$windows.access0(Ljava/lang/string;1) .
 如果已经配置了环境 还不行 在src新建包 org.apache.hadoop.io.nativeio
 然后hadoop02文件夹中的 NativeIO.java添加到这个包下 重新运行尝试

若要显示报错信息在resouces目录下添加log4j.properties
内容如下:

log4j.rootCategory=INFO,console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%dyy/MM/dd HH:mm:ss %p %c1: %m%n

分布式计算模型MapReduce的学习

1、mapreduce的定义

MapReduce是一个分布式运算程序的编程框架，是用户开发“基于Hadoop的数据分析应用”的核心框架。
MapReduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个Hadoop集群上

2、MapReduce编程模型

MapReduce是采用一种分而治之的思想设计出来的分布式计算框架
那什么是分而治之呢？
- 比如一复杂、计算量大、耗时长的的任务，暂且称为“大任务”；
- 此时使用单台服务器无法计算或较短时间内计算出结果时，可将此大任务切分成一个个小的任务，小任务分别在不同的服务器上并行的执行；
- 最终再汇总每个小任务的结果
MapReduce由两个阶段组成：
- Map阶段（切分成一个个小的任务）
- Reduce阶段（汇总小任务的结果）

1. Map阶段

map阶段有一个关键的map()函数；
此函数的输入是键值对
输出是一系列键值对，输出写入本地磁盘。

2. Reduce阶段

reduce阶段有一个关键的函数reduce()函数
此函数的输入也是键值对（即map的输出（kv对））
输出也是一系列键值对，结果最终写入HDFS

3、Map&Reduce

4、mapreduce编程指导思想

mapReduce编程模型的总结：
MapReduce的开发一共有八个步骤其中map阶段分为2个步骤，shuffle阶段4个步骤，reduce阶段分为2个步骤

1. Map阶段2个步骤

第一步：设置inputFormat类，将数据切分成key，value对，输入到第二步
第二步：自定义map逻辑，处理我们第一步的输入kv对数据，然后转换成新的key，value对进行输出

2. shuffle阶段4个步骤

第三步：对上一步输出的key，value对进行分区。（相同key的kv对属于同一分区）
第四步：对每个分区的数据按照key进行排序
第五步：对分区中的数据进行规约(combine操作)，降低数据的网络拷贝（可选步骤）
第六步：对排序后的kv对数据进行分组；分组的过程中，key相同的kv对为一组；将同一组的kv对的所有value放到一个集合当中（每组数据调用一次reduce方法）

3. reduce阶段2个步骤

第七步：对多个map的任务进行合并，排序，写reduce函数自己的逻辑，对输入的key，value对进行处理，转换成新的key，value对进行输出
第八步：设置outputformat将输出的key，value对数据保存到文件中

hadoop当中常用的数据类型

hadoop没有沿用java当中基本的数据类型，而是自己进行封装了一套数据类型，其自己封装的类型与java的类型对应如下
下表常用的数据类型对应的Hadoop数据序列化类型

Java类型	Hadoop Writable类型
Boolean	BooleanWritable
Byte	ByteWritable
Int	IntWritable
Float	FloatWritable
Long	LongWritable
Double	DoubleWritable
String	Text
Map	MapWritable
Array	ArrayWritable
byte[]	BytesWritable

3、词频统计例子

1、map方法，构建map逻辑

1、继承hadoop的Mapper类 Mapper<LongWritable, Text, Text, IntWritable>，后面四个是Hadoop的序列化类型，因为要涉及到网络传输，输入的key为LongWritable，表示行数，value表示一行的记录String类型，Text, IntWritable表示输出的String类型的key，IntWritable表示输出的value为Int类型。

2、重新父类的setup和map方法，setup主要是初始化操作，map则自定义key和value的逻辑操作

(避免了重复造对象的操作)。

3、自定义Map层的操作，将一行数据进行解析，并把单词设为key，每个单词value设为1.方便后面的统计操作。

 @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        //value=hello,hello
        String[] words = value.toString().split(",");
        for (String word : words) 
            keyOut.set(word);
            intWritable.set(1);
            context.write(keyOut,intWritable);

package wordCount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @program: bigData01
 * @ClassName WordCount
 * @description:
 * @author:蒋皓洁
 * @create: 2022-06-08 10:07
 * @Version 1.0
 **/
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> 


    private Text keyOut;
    private IntWritable intWritable;

    /**
     * 初始化的方法
     *
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException 
        keyOut = new Text();
        intWritable = new IntWritable();
    

    /**
     * 数据格式如下
     * hello,hello
     * world,world
     * hadoop,hadoop
     * 偏移量作为key，每一行的String作为Value
     * 读出则单词String类型作为key，数值作为value
     *
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        //value=hello,hello
        String[] words = value.toString().split(",");
        for (String word : words) 
            keyOut.set(word);
            intWritable.set(1);
            context.write(keyOut,intWritable);

2、reduce(进而合并)

1、reduce阶段继承 extends Reducer<Text, IntWritable, Text, IntWritable>，前面两个序列化的key为map的输出key和value，后面两个为reduce的输出key和value

2、重新两个方法setup和reduce，因为在shuffle阶段对主动对数据进行了分区，对分区中的数据进行规约(combine操作)，降低数据的网络拷贝（可选步骤）（这个步骤词频统计没有重新方法，自定义逻辑，用的默认方法）

3、setup初始化化数据操作

4、reduce方法操作，reduce的参数(Text key, Iterable<IntWritable> values, Context contex),前面两个表示map的key，和相同key的value的集合如下，后面的Context contex表示数据输出，

这个reduce表示对value值的累加操作得到intWritable值，输出context.write(key, intWritable);


     * (hadoop,1)
     * (hadoop,1)
     * (hadoop,1)
     * (spark,1)
     * hadoop,Iterable<IntWritable>(1,1,1)

  protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException 

        /**
         * 所以对值进行累加操作
         */
        int sum = 0;
        for (IntWritable value : values
        ) 
            sum += value.get();
        
        intWritable.set(sum);
        context.write(key, intWritable);

package wordCount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * @program: bigData01
 * @ClassName WordCountReduce
 * @description:
 * @author:蒋皓洁
 * @create: 2022-06-08 10:05
 * @Version 1.0
 **/
public class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> 


    private IntWritable intWritable;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException 
        intWritable = new IntWritable();
    

    /**
     * (hadoop,1)
     * (hadoop,1)
     * (hadoop,1)
     * (spark,1)
     * hadoop,Iterable<IntWritable>(1,1,1)
     *
     * @param key     单词
     * @param values  单词出现的次数组成的集合
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException 

        /**
         * 所以对值进行累加操作
         */
        int sum = 0;
        for (IntWritable value : values
        ) 
            sum += value.get();
        
        intWritable.set(sum);
        context.write(key, intWritable);

3、main方法

和八个步骤基本一致

package wordCount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


/**
 * @program: bigData01
 * @ClassName WordCountMapper
 * @description:
 * @author:蒋皓洁
 * @create: 2022-06-08 10:05
 * @Version 1.0
 **/
public class WordCount extends Configured implements Tool 


    @Override
    public int run(String[] args) throws Exception 
        Configuration configuration = super.getConf();


        //job对象
        Job job = Job.getInstance(configuration, WordCount.class.getName());
        //需要将程序提交到集群操作，需要setJarByClass
        job.setJarByClass(WordCount.class);
        //第一步:设置InputFormat，读取分片内容，生成key，v
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));// new Path(args[0])可写死，可传参
        //第二步：map逻辑，读取key，value生成k2 v2
        job.setMapperClass(WordCountMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        //第三，四，五，六分别是分区，排序，规约，分组
        //第七步:reduce逻辑
        job.setReducerClass(WordCountReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //第八步设置outPutFormat
        job.setOutputFormatClass(TextOutputFormat.class);
        //设置输出路径
        TextOutputFormat.setOutputPath(job, new Path(args[1]));

        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    

    public static void main(String[] args) throws Exception 
        Configuration entries = new Configuration();
        int run = ToolRunner.run(entries, new WordCount(), args);//传入三个参数
        System.exit(run);

4、join例子

订单数据表t_order：

id	date	pid	amount
1001	20150710	P0001	2
1002	20150710	P0002	3
1002	20150710	P0003	3

商品信息表t_product

id	pname	category_id	price
P0001	小米5	1000	2000
P0002	锤子T1	1000	3000

如果写SQL就是如下操作

select  a.id,a.date,b.name,b.category_id,b.price 
from t_order a 
join t_product  b
 on a.pid = b.id

1、map

map逻辑如下，查询到的，将商品表的ID作为key，方便后面计算

package joinCount;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;
/**
 * @program: bigData01
 * @ClassName ReduceJoinMapper
 * @description:
 * @author:蒋皓洁
 * @create: 2022-06-09 15:27
 * @Version 1.0
 **/
public class ReduceJoinMapper  extends Mapper<LongWritable, Text, Text, Text> 

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        //现在我们读取了两个文件，如何确定当前处理的这一行数据是来自哪一个文件里面的
        //方式一：通过获取文件的切片，获得文件明
        /*
        FileSplit inputSplit = (FileSplit) context.getInputSplit();//获取我们输入的文件的切片
        //获取文件名称
        String name = inputSplit.getPath().getName();
        if (name.equals("orders.txt")) 
            //订单表数据
         else 
            //商品表数据
        
        */

        String[] split = value.toString().split(",");

        //方式二：因为t_product表，都是以p开头，所以可以作为判断的依据
        if (value.toString().startsWith("p")) 
            //p0002,锤子T1,1000,3000
            //以商品id作为key2,相同商品的数据都会到一起去
            context.write(new Text(split[0]), value);
         else 
            //order
            // 1001,20150710,p0001,2
            context.write(new Text(split[2]), value);

2、reduce

对如果不是p开头的数据，则是订单数据，否则是商品基础信息数据

对数据进行筛选进而合并

package joinCount;

/**
 * @program: bigData01
 * @ClassName ReduceJoinReducer
 * @description:
 * @author:蒋皓洁
 * @create: 2022-06-09 15:47
 * @Version 1.0
 **/
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;

public class ReduceJoinReducer extends Reducer<Text, Text, Text, NullWritable> 
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException 
        //p0003 商品，订单的数据多条；保存订单信息
        ArrayList<String> orders = new ArrayList<>();
        //保存商品信息
        String product = "";

        for (Text value : values) 
            if (value.toString().startsWith("p")) //商品
                product = value.toString();
             else 
                orders.add(value.toString());
            
        

        for (String order : orders) 
            context.write(new Text(order + "\\t" + product), NullWritable.get());

3、main方法

main方法无多大区别

package joinCount;

/**
 * @program: bigData01
 * @ClassName ReduceJoinMain
 * @description:
 * @author:蒋皓洁
 * @create: 2022-06-09 15:57
 * @Version 1.0
 **/
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ReduceJoinMain extends Configured implements Tool 
    @Override
    public int run(String[] args) throws Exception 
        //获取job对象
        Job job = Job.getInstance(super.getConf(), ReduceJoinMain.class.getSimpleName());
        job.setJarByClass(ReduceJoinMain.class);

        //第一步：读取文件
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));

        //第二步：设置自定义mapper逻辑
        job.setMapperClass(ReduceJoinMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //分区，排序，规约，分组 省略

        //第七步：设置reduce逻辑
        job.setReducerClass(ReduceJoinReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //第八步：设置输出数据路径
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, new Path(args[1]));

        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    

    public static void main(String[] args) throws Exception 
        int run = ToolRunner.run(new Configuration(), new ReduceJoinMain(), args);
        System.exit(run);

5、map join

1、原理阐述

适用于关联表中有小表的情形；
可以将小表分发到所有的map节点，这样，map节点就可以在本地对自己所读到的大表数据进行join并输出最终结果，可以大大提高join操作的并发度，加快处理速度

2、实现示例

先在mapper类中预先定义好小表，进行join
引入实际场景中的解决方案：一次加载数据库或者用
定义mapper类：

因为订单表和商品是多对一的关系，所以我们可以先把商品表的数据预加载到map中处理

map如下

package joinMapCount;

/**
 * @program: bigData01
 * @ClassName MapJoinMapper
 * @description:
 * @author:蒋皓洁
 * @create: 2022-06-09 16:23
 * @Version 1.0
 **/
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
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;

public class MapJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> 
    //用于保存商品表的数据；productMap中的key是商品id，value是与key对应的表记录
    private Map<String, String> productMap;

    /**
     * 初始化方法，只在程序启动调用一次
     *
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException 

        productMap = new HashMap<String, String>();

        Configuration configuration = context.getConfiguration();

        //获取到所有的缓存文件
        //方式一
        URI[] cacheFiles = Job.getInstance(context.getConfiguration()).getCacheFiles();
        //方式二：deprecated
        //URI[] cacheFiles = DistributedCache.getCacheFiles(configuration);

        //现在只有一个缓存文件放进了分布式缓存中
        URI cacheFile = cacheFiles[0];

        //获取FileSystem
        FileSystem fileSystem = FileSystem.get(cacheFile, configuration);
        //读取文件，获取到输入流。这里面装的都是商品表的数据
        FSDataInputStream fsDataInputStream = fileSystem.open(new Path(cacheFile));

        /**
         * 商品表数据如下：
         * p0001,xiaomi,1000,2
         * p0002,appale,1000,3
         * p0003,samsung,1000,4
         */
        //获取到BufferedReader之后，可以一行一行的读取数据
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(fsDataInputStream));
        String line = null;
        //每次循环，获得表的一行数据
        while ((line = bufferedReader.readLine()) != null) 
            String[] split = line.split(",");
            productMap.put(split[0], line);
        
    

    /**
     * @param key
     * @param value   订单表的记录，如1001,20150710,p0001,2
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        String[] split = value.toString().split(",");
        //获取订单表的商品id
        String pid = split[2];

        //获取商品表的数据
        String pdtsLine = productMap.get(pid);

        context.write(new Text(value.toString() + "\\t" + pdtsLine), NullWritable.get());

main方法如下

package joinMapCount;

/**
 * @program: bigData01
 * @ClassName MapJoinMain
 * @description:
 * @author:蒋皓洁
 * @create: 2022-06-09 16:24
 * @Version 1.0
 **/
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.net.URI;

/**
 * 需求：同reduce join
 * 实现：select a.id, a.date, b.name, b.category_id, b.price
 * from t_order a
 * join t_product b
 * on a.pid = b.id
 */
public class MapJoinMain extends Configured implements Tool 
    @Override
    public int run(String[] args) throws Exception 
        //分布式缓存的hdfs路径
        URI uri = new URI("file:///D:/hadoop/mapjoin/cache/pdts.txt");
//
        //本地路径：需要用一下形式file:///C:/1、HK/.../pdts.txt   ；需要指定到具体的文件；如果是使用file:///c:\\\\1、HK\\\\3、ME...不符合语法会报错
        //URI uri = new URI("file:///C:/1、HK/3、ME/2、高级0x/1、Hadoop集群升级课件/9、MapReduce/MR第一次/12、join操作/map端join/cache/pdts.txt");
        Configuration configuration = super.getConf();
        //添加缓存文件 方式二：deprecated
        DistributedCache.addCacheFile(uri, configuration);

        //获取job对象
        Job job = Job.getInstance(configuration, MapJoinMain.class.getSimpleName());
        //添加缓存文件：方式一
//        job.addCacheFile(uri);
        job.setJarByClass(MapJoinMain.class);

        //读取文件，解析成为key，value对
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));

        job.setMapperClass(MapJoinMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        //没有reducer逻辑，不用设置了
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setNumReduceTasks(2);

        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    

    public static void main(String[] args) throws Exception 
        int run = ToolRunner.run(new Configuration(), new MapJoinMain(), args);
        System.exit(run);

6、自定义FileOutputFormat文件流输出

package goodOrBad;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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;

//泛型指的是输出的k，v类型
public class MyOutputFormat extends FileOutputFormat<Text, NullWritable> 

    @Override
    public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException 
        FileSystem fs = FileSystem.get(context.getConfiguration());
        Path goodComment = new Path("file:///D:\\\\hadoop\\\\5、自定义outputFormat\\\\测试传感器数据\\\\input\\\\good\\\\1.txt");
        Path badComment = new Path("file:///D:\\\\hadoop\\\\5、自定义outputFormat\\\\测试传感器数据\\\\input\\\\bad\\\\1.txt");
        FSDataOutputStream goodOutputStream = fs.create(goodComment);
        FSDataOutputStream badOutputStream = fs.create(badComment);
        return new MyRecordWriter(goodOutputStream, badOutputStream);
    

    static class MyRecordWriter extends RecordWriter<Text, NullWritable> 
        FSDataOutputStream goodStream = null;
        FSDataOutputStream badStream = null;

        public MyRecordWriter(FSDataOutputStream goodStream, FSDataOutputStream badStream) 
            this.goodStream = goodStream;
            this.badStream = badStream;
        

        @Override
        public void write(Text key, NullWritable value) throws IOException, InterruptedException 
            if (key.toString().split(",")[1].equals("\\"氧气传感器\\"")) //好评
                goodStream.write(key.toString().getBytes());
                goodStream.write("\\r\\n".getBytes());
             else //中评或差评
                badStream.write(key.toString().getBytes());
                badStream.write("\\r\\n".getBytes());
            
        

        //释放资源
        @Override
        public void close(TaskAttemptContext context) throws IOException, InterruptedException 
            if (badStream != null) 
                badStream.close();
            
            if (goodStream != null) 
                goodStream.close();

main方法如下：

package goodOrBad;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

public class MyOwnOutputFormatMain extends Configured implements Tool 
    @Override
    public int run(String[] args) throws Exception 
        Configuration conf = super.getConf();
        Job job = Job.getInstance(conf, MyOwnOutputFormatMain.class.getSimpleName());
        job.setJarByClass(MyOwnOutputFormatMain.class);

        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));

        job.setMapperClass(MyOwnMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        //使用默认的Reduce类的逻辑
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        job.setOutputFormatClass(MyOutputFormat.class);
        //设置一个输出目录，这个目录会输出一个success的成功标志的文件
        MyOutputFormat.setOutputPath(job, new Path(args[1]));

        //可以观察现象
        job.setNumReduceTasks(2);

        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    

    //kout：评分等级 0， 1， 2
    public static class MyOwnMapper extends Mapper<LongWritable, Text, Text, NullWritable> 
        @Override
        protected void map(LongWritable key, Text value, Mapper.Context context) throws IOException, InterruptedException 
            //评分
            //String commentStatus = split[9];
            context.write(value, NullWritable.get());
        
    

    public static void main(String[] args) throws Exception 
        Configuration configuration = new Configuration();
        ToolRunner.run(configuration, new MyOwnOutputFormatMain(), args);

路径参数args，在idea中的配置如下，参数以逗号隔开

放在liunx集群运行，参数之间也用空格隔开

hadoop jar mr01-1.0-SNAPSHOT.jar wordCount.WordCount /wordcount /wc010152

7、mapreduce的partitioner详解

在mapreduce执行当中，有一个默认的步骤就是partition分区；
- 分区主要的作用就是默认将key相同的kv对数据发送到同一个分区中；
- 在mapreduce当中有一个抽象类叫做Partitioner，默认使用的实现类是HashPartitioner，我们可以通过HashPartitioner的源码，查看到分区的逻辑如下
我们MR编程的第三步就是分区；这一步中决定了map生成的每个kv对，被分配到哪个分区里
- 那么这是如何做到的呢？
- 要实现此功能，涉及到了分区器的概念；

默认分区器HashPartitioner

MR框架有个默认的分区器HashPartitioner

我们能观察到：
- HashPartitioner实现了Partitioner接口
- 它实现了getPartition()方法
  - 此方法中对k取hash值
  - 再与MAX_VALUE按位与
  - 结果再模上reduce任务的个数
- 所以，能得出结论，相同的key会落入同一个分区中

自定义分区器

实际生产中，有时需要自定义分区的逻辑，让key落入我们想让它落入的分区
此时就需要自定义分区器
如何实现？
参考默认分区器HashPartitioner
- 自定义的分区器类，如CustomPartitioner
  - 实现接口Partitioner
  - 实现getPartition方法；此方法中定义分区的逻辑
- main方法
  - 将自定义的分区器逻辑添加进来job.setPartitionerClass(CustomPartitioner.class)
  - 设置对应的reduce任务个数job.setNumReduceTasks(3)

现有一份关于手机的流量数据，样本数据如下

格式说明

==需求==：使用mr，实现将不同的手机号的数据划分到6个不同的文件里面去，具体划分规则如下

135开头的手机号分到一个文件里面去，
136开头的手机号分到一个文件里面去，
137开头的手机号分到一个文件里面去，
138开头的手机号分到一个文件里面去，
139开头的手机号分到一个文件里面去，
其他开头的手机号分到一个文件里面去

根据mr编程8步，需要实现的代码有：

一、针对输入数据，设计JavaBean
二、自定义的Mapper逻辑（第二步）
三、自定义的分区类（第三步）
四、自定义的Reducer逻辑（第七步）
五、main程序入口

1、创建javaBean

实现Writable接口，重新接口的write和readFields方法，实现序列化和反序列化

package partitioner;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

/**
 * 实现Writable接口，重新接口的write和readFields方法，实现序列化和反序列化
 */
public class FlowBean implements Writable /*实现可序列化、反序列化*/ 
    private Integer upFlow;
    private Integer downFlow;
    private Integer upCountFlow;
    private Integer downCountFlow;

    /**
     * 序列化方法
     *
     * @param out
     * @throws IOException
     */
    @Override
    public void write(DataOutput out) throws IOException 
        out.writeInt(upFlow);
        out.writeInt(downFlow);
        out.writeInt(upCountFlow);
        out.writeInt(downCountFlow);
    


    /**
     * 反序列化方法
     * 反序列化需要注意：使用正确的反序列化的方法；反序列化的顺序跟序列化的顺序保持一致
     *
     * @param in
     * @throws IOException
     */

    @Override
    public void readFields(DataInput in) throws IOException 
        this.upFlow = in.readInt();
        this.downFlow = in.readInt();
        this.upCountFlow = in.readInt();
        this.downCountFlow = in.readInt();
    

    public Integer getUpFlow() 
        return upFlow;
    

    public void setUpFlow(Integer upFlow) 
        this.upFlow = upFlow;
    

    public Integer getDownFlow() 
        return downFlow;
    

    public void setDownFlow(Integer downFlow) 
        this.downFlow = downFlow;
    

    public Integer getUpCountFlow() 
        return upCountFlow;
    

    public void setUpCountFlow(Integer upCountFlow) 
        this.upCountFlow = upCountFlow;
    

    public Integer getDownCountFlow() 
        return downCountFlow;
    

    public void setDownCountFlow(Integer downCountFlow) 
        this.downCountFlow = downCountFlow;
    

    @Override
    public String toString() 
        return "FlowBean" +
                "upFlow=" + upFlow +
                ", downFlow=" + downFlow +
                ", upCountFlow=" + upCountFlow +
                ", downCountFlow=" + downCountFlow +
                '';

2、自定义map类

将手机号作为key，对象作为value，setup方法是初始化方法

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> 
    private FlowBean flowBean;
    private Text text;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException 
        flowBean = new FlowBean();
        text = new Text();
    

    /**
     * 1363157985066	13726230503	00-FD-07-A4-72-B8:CMCC
     * 120.196.100.82	i02.c.aliimg.com	游戏娱乐	24	27	2481	24681	200
     *
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        String[] fields = value.toString().split("\\t");

        String phoneNum = fields[1];
        String upFlow = fields[6];
        String downFlow = fields[7];
        String upCountFlow = fields[8];
        String downCountFlow = fields[9];

        text.set(phoneNum);

        flowBean.setUpFlow(Integer.parseInt(upFlow));
        flowBean.setDownFlow(Integer.parseInt(downFlow));
        flowBean.setUpCountFlow(Integer.parseInt(upCountFlow));
        flowBean.setDownCountFlow(Integer.parseInt(downCountFlow));

        context.write(text, flowBean);

3、继承Partitioner实现分区

package partitioner;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

/**
 * 继承Partitioner接口，实现分区
 */
public class PartitionOwn extends Partitioner<Text, FlowBean> 
    @Override
    public int getPartition(Text text, FlowBean flowBean, int numPartitions) 
        String phoenNum = text.toString();

        if(null != phoenNum && !phoenNum.equals(""))
            if(phoenNum.startsWith("135"))
                return 0;
            else if(phoenNum.startsWith("136"))
                return 1;
            else if(phoenNum.startsWith("137"))
                return 2;
            else if(phoenNum.startsWith("138"))
                return 3;
            else if(phoenNum.startsWith("139"))
                return 4;
            else 
                return 5;
            
        else
            return 5;

4、自定义Reducer

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class FlowReducer extends Reducer<Text, FlowBean, Text, Text> 

    /**
     * 同一个手机号的数据，调用一次reduce方法
     * @param key
     * @param values
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException 
        int upFlow = 0;
        int donwFlow = 0;
        int upCountFlow = 0;
        int downCountFlow = 0;

        for (FlowBean value : values) 
            upFlow += value.getUpFlow();
            donwFlow += value.getDownFlow();
            upCountFlow += value.getUpCountFlow();
            downCountFlow += value.getDownCountFlow();
        
        context.write(key, new Text(upFlow + "\\t" + donwFlow + "\\t" + upCountFlow + "\\t" + downCountFlow));

5、main方法

package partitioner;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class FlowMain extends Configured implements Tool 
    @Override
    public int run(String[] args) throws Exception 
        //获取job对象
        Job job = Job.getInstance(super.getConf(), "flowCount");
        //如果程序打包运行必须要设置这一句
        job.setJarByClass(FlowMain.class);

        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));

        job.setMapperClass(FlowMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        //自定义分区器：
        job.setPartitionerClass(PartitionOwn.class);
        //设置reduce个数
        job.setNumReduceTasks(Integer.parseInt(args[2]));

        job.setReducerClass(FlowReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, new Path(args[1]));

        boolean b = job.waitForCompletion(true);
        return b ? 0 : 1;
    


    public static void main(String[] args) throws Exception 
        Configuration configuration = new Configuration();
        //configuration.set("mapreduce.framework.name", "local");
        //configuration.set("yarn.resourcemanager.hostname", "local");

        int run = ToolRunner.run(configuration, new FlowMain(), args);
        System.exit(run);

注意：对于我们自定义分区的案例，==必须打成jar包上传到集群==上面去运行，因为我们本地已经没法通过多线程模拟本地程序运行了，将我们的数据上传到hdfs上面去，然后通过 hadoop jar提交到集群上面去运行，观察我们分区的个数与reduceTask个数的关系

8、mapreduce当中的排序

. 可排序的Key

排序是MapReduce框架中最重要的操作之一。
- MapTask和ReduceTask均会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。
- 默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。
对于MapTask，它会将处理的结果暂时放到环形缓冲区中，当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，并将这些有序数据溢写到磁盘上，而当数据处理完毕后，它会对磁盘上所有文件进行归并排序。
对于ReduceTask，它从每个执行完成的MapTask上远程拷贝相应的数据文件
- 如果文件大小超过一定阈值，则溢写磁盘上，否则存储在内存中。
- 如果磁盘上文件数目达到一定阈值，则进行一次归并排序以生成一个更大文件；
- 如果内存中文件大小或者数目超过一定阈值，则进行一次合并后将数据溢写到磁盘上。
- 当所有数据拷贝完毕后，ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。

2. 排序的种类：

1、部分排序

MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序
2、全排序

最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行架构
3、辅助排序

在Reduce端对key进行分组。应用于：在接收的key为bean对象时，想让一个或几个字段相同（全部字段比较不相同）的key进入到同一个reduce方法时，可以采用==分组排序==。
4、二次排序
- 二次排序：mr编程中，需要先按输入数据的某一列a排序，如果相同，再按另外一列b排序；
- mr自带的类型作为key无法满足需求，往往需要自定义JavaBean作为map输出的key
- JavaBean中，使用compareTo方法指定排序规则。

3. 二次排序

数据：样本数据如下；

每条数据有5个字段，分别是手机号、上行包总个数、下行包总个数、上行总流量、下行总流量
==需求==先对下行包总个数升序排序；若相等，再按上行总流量进行降序排序
根据mr编程8步，需要实现的代码有：
- 一、针对输入数据及二次排序规则，设计JavaBean
- 二、自定义的Mapper逻辑（第二步）
- 三、自定义的Reducer逻辑（第七步）
- 四、main程序入口
代码实现：
一、定义javaBean对象，用于封装数据及定义排序规则

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

//bean要能够可序列化且可比较，所以需要实现接口WritableComparable
public class FlowSortBean implements WritableComparable<FlowSortBean> 
    private String phone;
    //上行包个数
    private Integer upPackNum;
    //下行包个数
    private Integer downPackNum;
    //上行总流量
    private Integer upPayload;
    //下行总流量
    private Integer downPayload;

    //用于比较两个FlowSortBean对象

    /**
     * 先对下行包总个数升序排序；若相等，再按上行总流量进行降序排序
     * @param o
     * @return
     */
    @Override
    public int compareTo(FlowSortBean o) 
        //升序
        int i = this.downPackNum.compareTo(o.downPackNum);
        if (i == 0) 
            //降序
            i = -this.upPayload.compareTo(o.upPayload);
        
        return i;
    

    //序列化
    @Override
    public void write(DataOutput out) throws IOException 
        out.writeUTF(phone);
        out.writeInt(upPackNum);
        out.writeInt(downPackNum);
        out.writeInt(upPayload);
        out.writeInt(downPayload);
    

    //反序列化
    @Override
    public void readFields(DataInput in) throws IOException 
        this.phone = in.readUTF();
        this.upPackNum = in.readInt();
        this.downPackNum = in.readInt();
        this.upPayload = in.readInt();
        this.downPayload = in.readInt();
    

    @Override
    public String toString() 
        return phone + "\\t" + upPackNum + "\\t" + downPackNum + "\\t" + upPayload + "\\t" + downPayload;
    

    //setter、getter方法
    public String getPhone() 
        return phone;
    

    public void setPhone(String phone) 
        this.phone = phone;
    

    public Integer getUpPackNum() 
        return upPackNum;
    

    public void setUpPackNum(Integer upPackNum) 
        this.upPackNum = upPackNum;
    

    public Integer getDownPackNum() 
        return downPackNum;
    

    public void setDownPackNum(Integer downPackNum) 
        this.downPackNum = downPackNum;
    

    public Integer getUpPayload() 
        return upPayload;
    

    public void setUpPayload(Integer upPayload) 
        this.upPayload = upPayload;
    

    public Integer getDownPayload() 
        return downPayload;
    

    public void setDownPayload(Integer downPayload) 
        this.downPayload = downPayload;

二、自定义mapper类

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 FlowSortMapper extends Mapper<LongWritable, Text, FlowSortBean, NullWritable> 

    private FlowSortBean flowSortBean;

    //初始化
    @Override
    protected void setup(Context context) throws IOException, InterruptedException 
        flowSortBean = new FlowSortBean();
    

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        /**
         * 手机号	上行包	下行包	上行总流量	下行总流量
         * 13480253104	3	3	180	180
         */
        String[] fields = value.toString().split("\\t");

        flowSortBean.setPhone(fields[0]);
        flowSortBean.setUpPackNum(Integer.parseInt(fields[1]));
        flowSortBean.setDownPackNum(Integer.parseInt(fields[2]));
        flowSortBean.setUpPayload(Integer.parseInt(fields[3]));
        flowSortBean.setDownPayload(Integer.parseInt(fields[4]));

        context.write(flowSortBean, NullWritable.get());

三、自定义reducer类

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class FlowSortReducer extends Reducer<FlowSortBean, NullWritable, FlowSortBean, NullWritable> 

    @Override
    protected void reduce(FlowSortBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException 
        //经过排序后的数据，直接输出即可
        context.write(key, NullWritable.get());

四、main程序入口

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class FlowSortMain extends Configured implements Tool 
    @Override
    public int run(String[] args) throws Exception 
        //获取job对象
        Job job = Job.getInstance(super.getConf(), "flowSort");
        //如果程序打包运行必须要设置这一句
        job.setJarByClass(FlowSortMain.class);

        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path(args[0]));

        job.setMapperClass(FlowSortMapper.class);
        job.setMapOutputKeyClass(FlowSortBean.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setReducerClass(FlowSortReducer.class);
        job.setOutputKeyClass(FlowSortBean.class);
        job.setOutputValueClass(NullWritable.class);

        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, new Path(args[1]));

        boolean b = job.waitForCompletion(true);

        return b ? 0 : 1;
    


    public static void main(String[] args) throws Exception 
        Configuration configuration = new Configuration();
        int run = ToolRunner.run(configuration, new FlowSortMain(), args);
        System.exit(run);

9、mapreduce中的combiner

减少传输数据量，本地先combiner组合一下

1. combiner基本介绍

combiner类本质也是reduce聚合，combiner类继承Reducer父类
combine是运行在map端的，对map task的结果做聚合；而reduce是将来自不同的map task的数据做聚合
作用：
- combine可以减少map task落盘及向reduce task传输的数据量
是否可以做map端combine：
- 并非所有的mapreduce job都适用combine，无论适不适用combine，都不能对最终的结果造成影响；比如下边求平均值的例子，就不适用适用combine
```
Mapper 3 5 7 ->(3+5+7)/3=5 2 6 ->(2+6)/2=4 
 Reducer (3+5+7+2+6)/5=23/5   不等于   (5+4)/2=9/2
```

2. 需求：

对于我们前面的wordCount单词计数统计，我们加上Combiner过程，实现map端的数据进行汇总之后，再发送到reduce端，减少数据的网络拷贝
自定义combiner类

其实直接使用词频统计中的reducer类作为combine类即可
在main方法中加入

 job.setCombinerClass(MyReducer.class);

运行程序，观察控制台有combiner和没有combiner的异同
10、mapreduce中的GroupingComparator分组详解
关键类GroupingComparator
是mapreduce当中reduce端决定哪些数据作为一组，调用一次reduce的逻辑
默认是key相同的kv对，作为同一组；每组调用一次reduce方法；
可以自定义GroupingComparator，实现自定义的分组逻辑

1、自定义WritableComparator类

（1）继承WritableComparator
（2）重写compare()方法

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

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

protected OrderGroupingComparator() 
        super(OrderBean.class, true);

2. 需求：

现在有订单数据如下

订单id	商品id	成交金额
Order_0000001	Pdt_01	222.8
Order_0000001	Pdt_05	25.8
Order_0000002	Pdt_03	522.8
Order_0000002	Pdt_04	122.4
Order_0000002	Pdt_05	722.4
Order_0000003	Pdt_01	222.8

现在需要求取每个订单当中金额最大的商品
根据mr编程8步，需要实现的代码有：
- 一、针对输入数据及相同订单按金额降序排序，设计JavaBean
- 二、自定义的Mapper逻辑（第二步）
- 三、自定义分区器，相同订单分到同一区（第三步）
- 四、自定义分区内排序（在JavaBean中已完成）（第四步）
- 五、自定义分组，相同订单的为同一组（第六步）
- 六、自定义的Reducer逻辑（第七步）
- 七、main程序入口

1、JavaBean如下

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class OrderBean implements WritableComparable<OrderBean> 
    private String orderId;
    private Double price;

    /**
     * key间的比较规则
     *
     * @param o
     * @return
     */
    @Override
    public int compareTo(OrderBean o) 
        //注意：如果是不同的订单之间，金额不需要排序，没有可比性
        int orderIdCompare = this.orderId.compareTo(o.orderId);
        if (orderIdCompare == 0) 
            //比较金额，按照金额进行倒序排序
            int priceCompare = this.price.compareTo(o.price);
            return -priceCompare;
         else 
            //如果订单号不同，没有可比性，直接返回订单号的升序排序即可
            return orderIdCompare;
        
    

    /**
     * 序列化方法
     *
     * @param out
     * @throws IOException
     */
    @Override
    public void write(DataOutput out) throws IOException 
        out.writeUTF(orderId);
        out.writeDouble(price);
    

    /**
     * 反序列化方法
     *
     * @param in
     * @throws IOException
     */
    @Override
    public void readFields(DataInput in) throws IOException 
        this.orderId = in.readUTF();
        this.price = in.readDouble();
    

    public String getOrderId() 
        return orderId;
    

    public void setOrderId(String orderId) 
        this.orderId = orderId;
    

    public Double getPrice() 
        return price;
    

    public void setPrice(Double price) 
        this.price = price;
    

    @Override
    public String toString() 
        return orderId + "\\t" + price;

2、自定义mapper类：

mport 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 GroupMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> 

    /**
     * Order_0000001	Pdt_01	222.8
     * Order_0000001	Pdt_05	25.8
     * Order_0000002	Pdt_03	322.8
     * Order_0000002	Pdt_04	522.4
     * Order_0000002	Pdt_05	822.4
     * Order_0000003	Pdt_01	222.8
     * Order_0000003	Pdt_03	322.8
     * Order_0000003	Pdt_04	522.4
     *
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
        String[] fields = value.toString().split("\\t");
        //Order_0000003	Pdt_04	522.4
        OrderBean orderBean = new OrderBean();
        orderBean.setOrderId(fields[0]);
        orderBean.setPrice(Double.valueOf(fields[2]));

        //输出orderBean
        context.write(orderBean, NullWritable.get());

3、自定义分区类：

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Partitioner;

public class GroupPartitioner extends Partitioner<OrderBean, NullWritable> 
    @Override
    public int getPartition(OrderBean orderBean, NullWritable nullWritable, int numPartitions) 
        //将每个订单的所有的记录，传入到一个reduce当中
        return orderBean.getOrderId().hashCode() % numPartitions;

4、自定义分组类

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

//自定义分组类
public class MyGroup extends WritableComparator 
    public MyGroup() 
        //分组类：要对OrderBean类型的k进行分组
        super(OrderBean.class, true);
    

    @Override
    public int compare(WritableComparable a, WritableComparable b) 
        OrderBean a1 = (OrderBean) a;
        OrderBean b1 = (OrderBean) b;
        //需要将同一订单的kv作为一组
        return a1.getOrderId().compareTo(b1.getOrderId());

5、自定义reduce类

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class GroupReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> 

    /**
     * Order_0000002	Pdt_03	322.8
     * Order_0000002	Pdt_04	522.4
     * Order_0000002	Pdt_05	822.4
     * => 这一组中有3个kv
     * 并且是排序的
     * Order_0000002	Pdt_05	822.4
     * Order_0000002	Pdt_04	522.4
     * Order_0000002	Pdt_03	322.8
     *
     * @param key
     * @param values
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException 
        //Order_0000002	Pdt_05	822.4 获得了当前订单中进而最高的商品
        //top1
        context.write(key, NullWritable.get());

        //top2
        //这样出不了正确结果，只会将同一个key输出两次;结果如下
        /*
        *
            Order_0000001	222.8
            Order_0000001	222.8
            Order_0000002	822.4
            Order_0000002	822.4
            Order_0000003	222.8
            Order_0000003	222.8
        * */
//        for(int i = 0; i < 2; i++)
//            context.write(key, NullWritable.get());
//        

        //正确的做法：
//        int num = 0;
//        for(NullWritable value: values) 
//            context.write(key, value);
//            num++;
//            if(num == 2)
//                break;
//

6、自定义程序入口类

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 以上是关于MapReduce分布式计算的主要内容，如果未能解决你的问题，请参考以下文章 
 分布式计算模型MapReduce的学习
 MapReduce分布式计算
 MapReduce计算框架
 MapReduce(分布式计算)_01
 分布式计算 MapReduce与yarn工作机制
 大数据入门-分布式计算框架MapReduce

MapReduce分布式计算

MapReduce核心思想

MapReduce编程模型

词频统计

WorldCount代码编写

分布式计算模型MapReduce的学习

1、mapreduce的定义

2、MapReduce编程模型

1. Map阶段

2. Reduce阶段

3、Map&Reduce

4、mapreduce编程指导思想

hadoop当中常用的数据类型

3、 词频统计例子

1、map方法，构建map逻辑

2、reduce(进而合并)

3、main方法

4、join例子

1、map

2、reduce

3、main方法

5、map join

6、自定义FileOutputFormat文件流输出

7、mapreduce的partitioner详解

自定义分区器

8、mapreduce当中的排序

9、mapreduce中的combiner

10、mapreduce中的GroupingComparator分组详解

3、词频统计例子