使用MapReduce实现两个文件的Join操作

数据结构

customer表

1	hanmeimei	ShangHai	110
2	leilei	BeiJing	112
3	lucy	GuangZhou	119

oder表

1	1	50
2	1	200
3	3	15
4	3	350
5	3	58
6	1	42
7	1	352
8	2	1135
9	2	400
10	2	2000
11	2	300

MAPJOIN

场景：我们模拟一个有一份小表一个大表的场景，customer是那份小表，order是那份大表
做法：直接将较小的数据加载到内存中，按照连接的关键字建立索引，

大份数据作为MapTask的输入键值对 map()方法的每次输入都去内存当中直接去匹配连接。

然后把连接结果按 key 输出，这种方法要使用 hadoop中的 DistributedCache 把小份数据分布到各个计算节点，

每个 maptask 执行任务的节点都需要加载该数据到内存，并且按连接关键字建立索引。

环境配置：因为我们是在本地操作的，所以需要配置本地的hadoop

1。下载hadoop

2.解压到一个目录，记住，一会要用

配置电脑环境变量

如果你是一个初学者那么你就创建一个Java工程，步骤自己搜吧，网上一大堆然后创建一个mapjoin的包，在包里创建一个

JoinDemo的类然后如下第24行代码，在JoinDemo后加extened往后就根据我的代码敲就好了。

package mapjoin;
##以上内容标志了段代码是在mapjoin这个包里

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.*;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

##以上内容是需要导入的jar包，否则一些代码会报红。

public class JoinDemo extends Configured implements Tool {
    // customer文件在本地上的位置。
    // TODO: 改用参数传入
    private static final String CUSTOMER_CACHE_URL = "input/customer.txt"; --本地输入文件路径
//            "hdfs://hadoop1:9000/user/hadoop/mapreduce/cache/customer.txt";
    private static class CustomerBean {                 --
        private int custId;
        private String name;
        private String address;
        private String phone;

        public CustomerBean() {}

        public CustomerBean(int custId, String name, String address,String phone)
             {
            super();
            this.custId = custId;
            this.name = name;
            this.address = address;
            this.phone = phone;
        }



        public int getCustId() {
            return custId;
        }

        public String getName() {
            return name;
        }

        public String getAddress() {
            return address;
        }

        public String getPhone() {
            return phone;
        }
    }

    private static class CustOrderMapOutKey implements
            WritableComparable<CustOrderMapOutKey> {
        private int custId;
        private int orderId;

        public void set(int custId, int orderId) {
            this.custId = custId;
            this.orderId = orderId;
        }

        public int getCustId() {
            return custId;
        }

        public int getOrderId() {
            return orderId;
        }


        public void write( DataOutput out) throws IOException {
            out.writeInt(custId);
            out.writeInt(orderId);
        }


        public void readFields( DataInput in) throws IOException {
            custId = in.readInt();
            orderId = in.readInt();
        }

        public int compareTo(CustOrderMapOutKey o) {
            int res = custId -o.custId;
            return res == 0 ? orderId-o.orderId : res;
        }

        public boolean equals(Object obj) {
            if (obj instanceof CustOrderMapOutKey) {
                CustOrderMapOutKey o = (CustOrderMapOutKey)obj;
                return custId == o.custId && orderId == o.orderId;
            } else {
                return false;
            }
        }

        @Override
        public String toString() {
            return custId + "	" + orderId;
        }
    }

    private static class JoinMapper extends
            Mapper<LongWritable, Text, CustOrderMapOutKey, Text> {
        private final CustOrderMapOutKey outputKey = new CustOrderMapOutKey();
        private final Text outputValue = new Text();


        /**
         * 在内存中customer数据
         */
        private static final Map<Integer, CustomerBean> CUSTOMER_MAP = new HashMap<Integer, CustomerBean>();

        @Override
        protected void setup(Context context)
                throws IOException, InterruptedException {
            //读取缓存的文件
            FileSystem fs = FileSystem
                    .get( URI.create(CUSTOMER_CACHE_URL), context.getConfiguration());
            FSDataInputStream fdis = fs.open(new Path(CUSTOMER_CACHE_URL));

            BufferedReader reader = new BufferedReader(new InputStreamReader(fdis));
            String line = null;
            String[] cols = null;

            // 格式：客户编号  姓名  地址  电话
            while ((line = reader.readLine()) != null) {
                cols = line.split("	");
                if (cols.length < 4) {              // 数据格式不匹配，忽略
                    continue;
                }

                CustomerBean bean = new CustomerBean(Integer.parseInt(cols[0]), cols[1], cols[2], cols[3]);
                //缓存客户信息
                CUSTOMER_MAP.put(bean.getCustId(), bean);
            }
        }

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

            // 格式: 订单编号 客户编号    订单金额
            String[] cols = value.toString().split("	");
            if (cols.length < 3) {
                return;
            }

            int custId = Integer.parseInt(cols[1]);     // 取出客户编号
            CustomerBean customerBean = CUSTOMER_MAP.get(custId);

            if (customerBean == null) {         // 没有对应的customer信息可以连接
                return;
            }

            StringBuffer sb = new StringBuffer();
            sb.append(cols[2])
                    .append("	")
                    .append(customerBean.getName())
                    .append("	")
                    .append(customerBean.getAddress())
                    .append("	")
                    .append(customerBean.getPhone());

            outputValue.set(sb.toString());
            outputKey.set(custId, Integer.parseInt(cols[0]));

            //WC 操作
//            context.write(value, 1);

            context.write(outputKey, outputValue);
        }

    }

    /**
     * reduce
     * @author Ivan
     *
     */
    private static class JoinReducer extends
            Reducer<CustOrderMapOutKey, Text, CustOrderMapOutKey, Text> {
        @Override
        protected void reduce(CustOrderMapOutKey key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            // 什么事都不用做，直接输出
            for (Text value : values) {
                context.write(key, value);
            }
        }
    }
    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
//        if (args.length < 2) {
//            new IllegalArgumentException("Usage: <inpath> <outpath>");
//            return;
//        }
        //调用 JoinDemo 对象的run方法
        ToolRunner.run(new Configuration(), new JoinDemo(), args);
    }


    public int run(String[] args) throws Exception {
        System.setProperty("hadoop.home.dir", "D:\software\hadoop-2.6.0\hadoop-2.6.0");
        //默认配置
        Configuration conf = getConf();
        FileSystem fileSystem=FileSystem.get(conf);
        Path outputPath=new Path("output/");
        if(fileSystem.exists(outputPath)){
            fileSystem.delete(outputPath,true);
        }

        Job job = Job.getInstance(conf);
        job.setJarByClass(JoinDemo.class);

        // 添加customer cache文件
        job.addCacheFile(URI.create(CUSTOMER_CACHE_URL));
        Path inputPath=new Path("input/order.txt");
//        Path inputPath= new Path(args[0])
        FileInputFormat.addInputPath(job, inputPath);

//        Path outputPath=new Path(args[1]);
        FileOutputFormat.setOutputPath(job, outputPath);

        // map settings
        job.setMapperClass(JoinMapper.class);
        job.setMapOutputKeyClass(CustOrderMapOutKey.class);
        job.setMapOutputValueClass(Text.class);

        // reduce settings
//        job.setReducerClass(JoinReducer.class);
//        job.setOutputKeyClass(CustOrderMapOutKey.class);
//        job.setOutputKeyClass(Text.class);
        boolean res = job.waitForCompletion(true);
        return res ? 0 : 1;
    }
}

 本地的文件位置

 最后执行结果：

1	1	hanmeimei	ShangHai	50	110
2	1	hanmeimei	ShangHai	200	110
3	3	lucy	GuangZhou	15	119
4	3	lucy	GuangZhou	350	119
5	3	lucy	GuangZhou	58	119
6	1	hanmeimei	ShangHai	42	110
7	1	hanmeimei	ShangHai	352	110
8	2	leilei	BeiJing	1135	112
9	2	leilei	BeiJing	400	112
10	2	leilei	BeiJing	2000	112
11	2	leilei	BeiJing	300	112