离线计算组件篇-MapReduce-Join连接

11.mapreduce的join

1. reduce join

(1)、需求

订单数据表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

假如数据量巨大，两表的数据是以文件的形式存储在HDFS中，需要用mapreduce程序来实现一下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  

(2)、实现机制

• 通过将关联的条件作为map输出的key，将两表满足join条件的数据并携带数据所来源的文件信息，发往同一个reduce task，在reduce中进行数据的串联
• 定义Mapper：

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;

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);
        }
    }
}

• 定义Reducer：

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

import java.io.IOException;

public class ReduceJoinReducer extends Reducer<Text,Text,Text, NullWritable> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        String order = "";
        String product = "";
        for (Text value : values) {
            if (value.toString().startsWith("p")) {
                product = value.toString();
            }else {
                order = value.toString();
            }
        }
        // 将结果保存
        context.write(new Text(order + "\t" + product),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.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);
    }
}

2. map join

(1)、原理阐述

• 适用于关联表中有小表的情形；

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

• 将小文件上传到hdfs中，然后在Main函数中将文件添加到缓存中，然后使用java io的方式读取文件；

  在Map()函数中将要实现join的字段放在key中，从而实现两个文件之间的join。

(2)、实现示例

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

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>();
        // 通过context上下文对象去 获取配置文件对象
        Configuration configuration = context.getConfiguration();

        // 通过Job对象去 获取到所有的缓存文件
        // 方式一
        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中的key是商品id，value是与key对应的表记录
            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 productLine = productMap.get(pid);

        System.out.println("Value中的数据:" + value.toString());
        System.out.println("商品表中的数据:" + productLine);
        // 此时把所有的数据直接拼接在一起了，然后便于观察最后的实际拼接的结果;在实际条件下其实也可以使用这种方式，然后在测试的没有问题之后的,可以再把这个删除
        
        context.write(new Text(value.toString() + "\t" + productLine), NullWritable.get());
    }
}

• 定义main方法：

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("hdfs://node01:8020/cache/product.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);
    }
}