hadoop之join链接

1、数据量大;

import org.apache.hadoop.conf.Configuration;
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 App {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "file:///");
        conf.set("mapreduce.framework.name", "local");

        Job job = Job.getInstance(conf);
        job.setJobName("ReduceJoinApp");
        job.setJarByClass(App.class);
        //
        FileInputFormat.addInputPath(job, new Path("d:/mr/join"));
        FileOutputFormat.setOutputPath(job, new Path("d:/mr/join/out"));
        //
        job.setMapperClass(ReduceJoinMapper.class);
        job.setReducerClass(ReduceJoinReducer.class);

        job.setMapOutputKeyClass(ComboKey.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //分区类
        job.setPartitionerClass(CIDPartitioner.class);
        //排序类
        job.setSortComparatorClass(ComboKeyComparator.class);
        //分组类
        job.setGroupingComparatorClass(CIDGroupComparator.class);

        job.waitForCompletion(true);
    }
}



import org.apache.hadoop.io.WritableComparable;

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

/**
 * 组合可以
 */
public class ComboKey implements WritableComparable<ComboKey> {
    //0-customer 1-order
    public int type ;
    //cid
    public int cid ;
    //oid
    public int oid ;

    public int compareTo(ComboKey o) {
        int type0 = o.type ;
        int cid0 = o.cid ;
        int oid0 = o.oid ;
        //customer
        if(type == 0){
            //customer
            if(type0 == 0){
                return cid - cid0 ;
            }
            //order
            else{
                if(cid == cid0){
                    return -1;
                }
                else{
                    return cid - cid0 ;
                }
            }
        }
        //order
        else{
            //customer
            if(type0 == 0){
                if(cid == cid0){
                    return 1 ;
                }
                else{
                    return cid - cid0 ;
                }
            }
            //orders
            else{
                if(cid == cid0){
                    return oid - oid0 ;
                }
                else{
                    return cid - cid0 ;
                }
            }
        }
    }

    public void write(DataOutput out) throws IOException {
        out.writeInt(type);
        out.writeInt(cid);
        out.writeInt(oid);
    }

    public void readFields(DataInput in) throws IOException {
        this.type = in.readInt();
        this.cid = in.readInt();
        this.oid = in.readInt();
    }
}



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

/**
 *
 */
public class ComboKeyComparator extends WritableComparator {
    protected ComboKeyComparator() {
        super(ComboKey.class, true);
    }

    public int compare(WritableComparable w1, WritableComparable w2) {
        ComboKey k1 = (ComboKey) w1;
        ComboKey k2 = (ComboKey) w2;
        return k1.compareTo(k2);
    }
}



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


/**
 * Created by Administrator on 2017/6/1.
 */
public class CIDPartitioner extends Partitioner<ComboKey,Text>{
    public int getPartition(ComboKey key, Text text, int numPartitions) {
        return key.cid % numPartitions;
    }
}


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

/**
 * Created by Administrator on 2017/6/1.
 */
public class CIDGroupComparator extends WritableComparator {
    protected CIDGroupComparator() {
        super(ComboKey.class, true);
    }

    public int compare(WritableComparable w1, WritableComparable w2) {
        ComboKey k1 = (ComboKey) w1;
        ComboKey k2 = (ComboKey) w2;
        return k1.cid - k2.cid ;
    }
}


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,ComboKey, Text>{
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] arr = line.split(",") ;
        FileSplit split = (FileSplit) context.getInputSplit();
        ComboKey k = new ComboKey();
        //customers
        if(split.getPath().toString().contains("customers")){
            k.type = 0 ;
            k.cid = Integer.parseInt(arr[0]);
        }
        //orders
        else{
            k.type= 1 ;
            k.cid = Integer.parseInt(arr[3]);//cid
            k.oid = Integer.parseInt(arr[0]);//oid
        }
        context.write(k, value);
    }
}


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;


/**
 * Created by Administrator on 2017/6/1.
 */
public class ReduceJoinReducer extends Reducer<ComboKey,Text,Text,NullWritable> {

    protected void reduce(ComboKey key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        Iterator<Text> it = values.iterator();
        int type = key.type ;
        String custInfo = null ;
        //customer
        if(type == 0){
            Text custText = it.next();
            custInfo = custText.toString() ;
            context.write(new Text(custInfo), NullWritable.get());
            while(it.hasNext()){
                String orderInfo = it.next().toString();
                context.write(new Text(custInfo + "," + orderInfo), NullWritable.get());
            }
        }
        //订单(没有客户)
        else{
            while(it.hasNext()){
                String orderInfo = it.next().toString();
                context.write(new Text("null," + orderInfo),NullWritable.get());
            }
        }
    }
}

2、数据量小，可直接内存加载

import com.it18zhang.hadoop.mapreduce.inputformat.wholefile.WholeFileInputFormat;
import com.it18zhang.hadoop.mapreduce.inputformat.wholefile.WordCountMapper;
import com.it18zhang.hadoop.mapreduce.inputformat.wholefile.WordCountReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

import java.io.IOException;

/**
 * Created by Administrator on 2017/6/1.
 */
public class App {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "file:///");
        conf.set("mapreduce.framework.name", "local");
        conf.set("customers.data.path", "file:///d:/mr/customers.txt");
        //创建job
        Job job = Job.getInstance(conf);
        job.setJobName("MapJoin");
        job.setJarByClass(App.class);

        //添加输入路径(可以多次添加)
        FileInputFormat.addInputPath(job, new Path("d:/mr/orders.txt"));
        //设置输出路径，只能一个。而且目录不能存在。
        FileOutputFormat.setOutputPath(job, new Path("d:/mr/out"));

        //设置mapper类
        job.setMapperClass(JoinMapper.class);
        job.setNumReduceTasks(0);

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

        job.waitForCompletion(true);
    }
    
}

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.Text;
import org.apache.hadoop.io.NullWritable;
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;


/**
 *
 */
public class JoinMapper extends Mapper<LongWritable,Text,Text,NullWritable>{
    //存放所有customers
    private Map<String,String> allCustomers = new HashMap<String, String>();

    /**
     * 读取customers数据到内存中
     */
    protected void setup(Context context) throws IOException, InterruptedException {
        //得到文件系统
        Configuration conf = context.getConfiguration();
        FileSystem fs = FileSystem.get(conf);
        String path = conf.get("customers.data.path") ;
        FSDataInputStream in = fs.open(new Path(path));
        BufferedReader reader = new BufferedReader(new InputStreamReader(in));
        String line = null ;
        while((line = reader.readLine()) != null){
            String cid = line.split(",")[0] ;
            allCustomers.put(cid,line);
        }
        reader.close();
    }

    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] arr = line.split(",");
        //cid
        String cid = arr[3] ;
        String custInfo = allCustomers.get(cid);
        context.write(new Text(custInfo + "," + line),NullWritable.get());
    }
}