自定義一個mapper類需要實現(xiàn)如下步驟

package cn.leon.reduce_join;

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.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;

public class MapperJoinMapper extends Mapper<LongWritable, Text,Text,Text> {

    private HashMap<String, String> stringHashMap = new HashMap<>();

    //第一件事情:將分布式緩存的小表數(shù)據(jù)讀取到本地Map集合(只需要做一次)

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //1:獲取分布式緩存文件列表
        URI[] cacheFiles =  context.getCacheFiles();

        //2:獲取指定的分布式緩存文件的文件系統(tǒng)(FileSystem)
        FileSystem fileSystem = FileSystem.get(cacheFiles[0], context.getConfiguration());

        //3:獲取文件的輸入流
        FSDataInputStream inputStream = fileSystem.open(new Path(cacheFiles[0]));

        //4:讀取文件內(nèi)容, 并將數(shù)據(jù)存入Map集合
        //4.1 將字節(jié)輸入流轉(zhuǎn)為字符緩沖流FSDataInputStream --->BufferedReader
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
        //4.2 讀取小表文件內(nèi)容,以行位單位,并將讀取的數(shù)據(jù)存入map集合


        String line = null;
        while((line = bufferedReader.readLine()) != null){
            String[] split = line.split(",");

            stringHashMap.put(split[0], line);

        }
        //5:關(guān)閉流
        bufferedReader.close();
        fileSystem.close();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] split = value.toString().split(",");
        String productId = split[2];

        //k2
        String productLine = stringHashMap.get(productId);
        //v2
        String valueLine = productLine + "\t" + value.toString();


        context.write(new Text(productId),new Text(valueLine));
    }
}

主類中

package cn.leon.reduce_join;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
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;

import java.io.File;
import java.net.URI;

public class ReducerMain extends Configured implements Tool {
    @Override
    public int run(String[] strings) throws Exception {

        Job job = Job.getInstance(super.getConf(),"reducer_join");

        //把小表放在分布式緩存中
        job.addCacheFile(new URI("hdfs://node01:8020/input/join/product.txt"));

        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job,new Path("/Users/caoxiaozhu/Desktop/reduce"));

        //分布式緩存mapper
        job.setMapperClass(MapperJoinMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        Path path = new Path("/Users/caoxiaozhu/Desktop/result");
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job,path);

        FileSystem fileSystem = FileSystem.get(new URI("/Users/caoxiaozhu/Desktop/result"),new Configuration());
        if (fileSystem.exists(path)){
            fileSystem.delete(path,true);
        }

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

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

相當(dāng)于不需要reducer，直接在mapper中合并k2 v2即可。