Class Map<LongWritable, Text, LongWritable, Text>{method map(){// 获取文件的每一行数据，并以":"分割String[] line = value.toString().split(":");//split对应的文件名String fileName = ((FileSplit) context.getInputSplit()).getPath().getName();//处理company文件的value信息："Beijing Red Star:1"if (path.indexOf("company") >= 0){//<key,value> --<"1","company:Beijing Red Star">context.write(new LongWritable(line[1]), new Text("company" + ":" + line[0]));}//处理adress文件的value信息："1:Beijing"else if (path.indexOf("address") >= 0){//<key,value> --<"1","address:Beijing">context.write(new LongWritable(line[0]), new Text("address" + ":" + line[1]));}}
}

Class Reducer<LongWritable, Text, Text, Text>{method reduce(){//用来存储 company 和 address 的集合List<String> companys = new ArrayList<String>();List<String> addresses = new ArrayList<String>();for(Text text : v2s){            String[] result = text.toString().split(":");//以 company 开头的value存储到 company 集合中if(result[0].equals("company")){companys.add(result[1]);}//以 address 开头的value存储到 address 集合中else if(result[0].equals("address")){addresses.add(result[1]);}}        /**         * 如果只判断左表addresses.size()!=0;则转化为 左外连接 --> LEFT OUTER JOIN         * 如果只判断右表companys.size()!=0;则转化为 右外连接 --> RIGHT OUTER JOIN         * 左右都不判断,则转化为 全外连接 --> FULL OUTER JOIN         */        // 求笛卡尔积if(0 != companys.size()&& 0 != addresses.size()){for(int i=0;i<companys.size();i++){for(int j=0;j<addresses.size();j++){//<key,value>--<"Beijing JD","Beijing">context.write(new Text(companys.get(i)), new Text(addresses.get(j)));}}}}
}