Hadoop基础-MapReduce的Join操作

　　　　　　　　　　　　　　　　　　Hadoop基础-MapReduce的Join操作

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　作者：尹正杰

版权声明：原创作品，谢绝转载！否则将追究法律责任。

一.连接操作Map端Join（适合处理小表+大表的情况）

1    no001    12.3    7
2    no002    18.8    4
3    no003    20.0    3
4    no004    50.0    7
5    no005    23.1    2
6    no006    39.0    3
7    no007    5.0    2
8    no008    6.0    1

1    linghunbaiduren
2    yinzhengjie
3    alex
4    linhaifeng
5    wupeiqi
6    xupeicheng
7    changqiling
8    laowang

1>.MapJoinMapper.java 文件内容

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.join.map;

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.IntWritable;
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.util.HashMap;
import java.util.Map;


/**
 * 输出KeyValue
 * key是组合后的数据
 * value空
 *
 */
public class MapJoinMapper extends Mapper<LongWritable,Text,Text,NullWritable> {

    Map<Integer,String> map = new HashMap<Integer, String>();

    
    /**
    *
    *setup方法是在map方法之前执行，它也是map方法的初始化操作.
    *
    */
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //通过上下文，得到conf
        Configuration conf = context.getConfiguration();
        //通过conf获取自定义key
        String file = conf.get("customer.file");
        //读取customer数据
        FileSystem fs = FileSystem.get(conf);
        FSDataInputStream fis = fs.open(new Path(file));
        InputStreamReader reader = new InputStreamReader(fis);
        BufferedReader br = new BufferedReader(reader);
        String line = null;
        byte[] buf = new byte[1024];
        while((line = br.readLine()) != null){
            String[] arr = line.split("	");
            int id = Integer.parseInt(arr[0]);
            String name = arr[1];
            //1 tom
            //2 tomas
            map.put(id,name);
        }
    }

    /**
     * 通过
     * oid  orderno price   cid
     * 8    no008    6.0        1
     *
     * 得到
     * cid  cname   orderno price
     * 1    tom     no008   6.0
     */

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String line = value.toString();

        String[] arr = line.split("	");

        String orderno = arr[1];
        String price = arr[2];
        int cid = Integer.parseInt(arr[3]);

        String name = map.get(cid);
        //拼串操作
        String outKey = cid + "	" + name + "	" + orderno + "	" + price + "	";
        //
        context.write(new Text(outKey), NullWritable.get());
    }
}

2>.MapJoinApp.java 文件内容

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.join.map;

import org.apache.hadoop.conf.Configuration;
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.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MapJoinApp {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        //自定义一个变量名"customer.file"，后面的文件是其具体的值，这里设置后可以在Mapper端通过get方法获取改变量的值。
        conf.set("customer.file", "D:\10.Java\IDE\yhinzhengjieData\customers.txt");
        conf.set("fs.defaultFS","file:///");
        FileSystem fs = FileSystem.get(conf);
        Job job = Job.getInstance(conf);
        job.setJarByClass(MapJoinApp.class);
        job.setJobName("Map-Join");
        job.setMapperClass(MapJoinMapper.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        FileInputFormat.addInputPath(job,new Path("D:\10.Java\IDE\yhinzhengjieData\orders.txt"));
        Path outPath = new Path("D:\10.Java\IDE\yhinzhengjieData\out");
        if (fs.exists(outPath)){
            fs.delete(outPath);
        }
        FileOutputFormat.setOutputPath(job,outPath);
        job.waitForCompletion(true);
    }
}

3>.验证结果是否正确

二.连接操作Reduce端Join之组合Key实现（适合处理大表+大表的情况）

1    no001    12.3    7
2    no002    18.8    4
3    no003    20.0    3
4    no004    50.0    7
5    no005    23.1    2
6    no006    39.0    3
7    no007    5.0    2
8    no008    6.0    1

1    linghunbaiduren
2    yinzhengjie
3    alex
4    linhaifeng
5    wupeiqi
6    xupeicheng
7    changqiling
8    laowang

　　以上两个文件的指定路径如下：（输入路径）

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.join.reduce;

import org.apache.hadoop.io.WritableComparable;

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

public class CompKey implements WritableComparable<CompKey> {
    //定义客户id
    private int cid;
    //定义标识
    private int flag;

    public int compareTo(CompKey o) {
        //如果cid相等
        if (this.getCid() == o.getCid()) {
            //比较flag
            return this.getFlag() - o.getFlag();
        }
        return this.getCid() - o.getCid();
    }

    //定义序列化
    public void write(DataOutput out) throws IOException {
        out.writeInt(cid);
        out.writeInt(flag);
    }

    //定义反序列化
    public void readFields(DataInput in) throws IOException {
        cid = in.readInt();
        flag = in.readInt();
    }

    public int getCid() {
        return cid;
    }

    public void setCid(int cid) {
        this.cid = cid;
    }

    public int getFlag() {
        return flag;
    }

    public void setFlag(int flag) {
        this.flag = flag;
    }

    @Override
    public String toString() {
        return cid + "," + flag;
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.join.reduce;

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

public class MyGroupingComparator extends WritableComparator {

    public MyGroupingComparator() {
        super(CompKey.class, true);
    }

    @Override
    public int compare(WritableComparable a, WritableComparable b) {

        CompKey ck1 = (CompKey) a;
        CompKey ck2 = (CompKey) b;

        int cid1 = ck1.getCid();
        int cid2 = ck2.getCid();


        return cid1 - cid2;
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.join.reduce;

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;

public class ReduceJoinMapper extends Mapper<LongWritable, Text, CompKey, Text> {

    String fileName;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //得到输入切片
        InputSplit split = context.getInputSplit();
        FileSplit fileSplit = (FileSplit) split;

        //得到切片文件名或路径
        fileName = fileSplit.getPath().getName();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

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

            //判断文件是否包含"customers"。
            if (fileName.contains("customers")) {
                int cid = Integer.parseInt(arr[0]);
                CompKey ck = new CompKey();
                ck.setCid(cid);
                ck.setFlag(0);
                context.write(ck, value);
            } else {
                int cid = Integer.parseInt(arr[3]);
                CompKey ck = new CompKey();
                ck.setCid(cid);
                ck.setFlag(1);
                context.write(ck, value);
        }
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.join.reduce;

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.Iterator;

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


    /**
     * 通过
     * oid  orderno price   cid
     * 8    no008    6.0        1
     * <p>
     * 得到
     * cid  cname   orderno price
     * 1    tom     no008   6.0
     */
    @Override
    protected void reduce(CompKey key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

        //初始化迭代器
        Iterator<Text> it = values.iterator();

        //将while指针指向第一条之后
        String cust = it.next().toString();

        //继上一条之后读取
        while(it.hasNext()){
            String[] arr = it.next().toString().split("	");
            String orderno = arr[1];
            String price = arr[2];
            String newLine = cust.toString() + "	" + orderno + "	" + price;
            context.write(new Text(newLine), NullWritable.get());

        }
    }
}

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Hadoop%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:y1053419035@qq.com
*/
package cn.org.yinzhengjie.join.reduce;

import org.apache.hadoop.conf.Configuration;
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.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class ReduceJoinApp {


    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","file:///");
        FileSystem fs = FileSystem.get(conf);
        Job job = Job.getInstance(conf);
        job.setJarByClass(ReduceJoinApp.class);
        job.setJobName("Reduce-Join");
        job.setMapperClass(ReduceJoinMapper.class);
        job.setReducerClass(ReduceJoinReducer.class);
        job.setGroupingComparatorClass(MyGroupingComparator.class);
        //map的输出k-v
        job.setMapOutputKeyClass(CompKey.class);
        job.setMapOutputValueClass(Text.class);

        //reduce的k-v
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //指定输入的文件路径
        FileInputFormat.addInputPath(job,new Path("D:\10.Java\IDE\yhinzhengjieData\input\"));
        //指定输出的文件路径
        Path outPath = new Path("D:\10.Java\IDE\yhinzhengjieData\output");
        if (fs.exists(outPath)){
            fs.delete(outPath);
        }
        FileOutputFormat.setOutputPath(job,outPath);

        job.setNumReduceTasks(2);
        job.waitForCompletion(true);
    }
}

　　以上代码执行结果如下：（输出路径）