map侧连接

两个数据集中一个非常小，可以让小数据集存入缓存。在作业开始这些文件会被复制到运行task的节点上。 一开始，它的setup方法会检索缓存文件。

与reduce侧连接不同，Map侧连接需要等待参与连接的数据集满足如下条件:

1.除了连接键外，所有的输入都必须按照连接键排序。 输入的各种数据集必须有相同的分区数。 所有具有相同键的记录需要放在同一分区中。 当Map任务对其他Mapreduce作业的结果进行处理时（Cleanup时），Map侧的连接条件都自动满足 CompositeInputFormat类用于执行Map侧的连接，而输入和连接类型的配置可以通过属性指定。

2.如果其中一个数据集足够小，旁路的分布式通道可以用在Map侧的连接中。

实例：

输入：　　　

   123（工厂）　　　　　　　　　　　　　　　　　　　　　　　a（地址表）：

Beijing Red Star,1　　　　　　　　　　　　　　　　　　　　　　1,Beijing

Shenzhen Thunder,3　　　　　　　　　　　　　　　　　　　　   2,Guangzhou

Guangzhou Honda,2　　　　　　　　　　　　　　　　　　　　　3,Shenzhen

Beijing Rising,1　　　　　　　　　　　　　　　　　　　　　　　 4,xian　　　

Guangzhou Development Bank,2

Tencent,3

Back of Beijing,1

思路：在map端中的cache载入地址表，在map阶段的setup()中，定义HashMap()，将字符串分割，放入HashMap中，然后在map阶段，利用hashmap。get（），得到对应的地址。

代码：

package mapreduce01;

import java.io.IOException;

import java.net.URI;

import java.util.HashMap;

import java.util.Map;

import org.apache.hadoop.conf.Configuration;

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

public class Mapduan {

static String INPUT_PATH = "hdfs://master:9000/qq/123";

static String OUTPUT_PATH="hdfs://master:9000/output";

static class MyMapper extends Mapper<Object,Object,Text,Text>{

Text output_key = new Text();

Text output_value = new Text();

Map<String,String> addMap = new HashMap<String,String>();   //image  yingshe

protected void setup(Context context) throws java.io.IOException, java.lang.InterruptedException{

URI uri=context.getCacheFiles()[0];

Path path = new Path(uri);

FileSystem fs = path.getFileSystem(context.getConfiguration());

LineReader lineReader = new LineReader(fs.open(path));

Text line=new Text();

while(lineReader.readLine(line)>0){

String tokens[] = line.toString().split(",");

if(tokens!=null && tokens.length==2)

     addMap.put(tokens[0], tokens[1]);                

}

}

protected void map(Object key,Object value,Context context) throws IOException,InterruptedException{

String[] tokens = value.toString().split(",");

if(tokens!=null&&tokens.length==2){

output_key.set(tokens[0]);

String addrName = addMap.get(tokens[1].toString());

output_value.set(addrName);

context.write(output_key,output_value);

}

}

}

static class MyReduce extends Reducer<Text,Text,Text,Text> {

Text  output_key=new Text();

Text  output_value=new Text();

protected void reduce(Text key, Iterable<Text> values,Context context)  throws IOException,InterruptedException{

context.write(key,values.iterator().next());

}

}

public static void main(String[] args) throws Exception{

Path outputpath = new Path(OUTPUT_PATH);

Path cacheFile = new Path("hdfs://master:9000/qq/a");

Configuration conf = new Configuration();

FileSystem fs = outputpath.getFileSystem(conf);

if(fs.exists(outputpath)){

fs.delete(outputpath,true);

}

 Job  job=Job.getInstance(conf);

 FileInputFormat.setInputPaths(job,INPUT_PATH);

 FileOutputFormat.setOutputPath(job, outputpath);

 URI uri =cacheFile.toUri();

 job.setCacheFiles(new URI[]{uri});  //set cache address

 job.setMapperClass(MyMapper.class);

 job.setReducerClass(MyReduce.class);   

 job.setOutputKeyClass(Text.class);

 job.setOutputValueClass(Text.class);

 job.waitForCompletion(true);

}

}

实验结果：

Back of Beijing Beijing

Beijing Red Star Beijing

Beijing Rising Beijing

Guangzhou Development Bank Guangzhou

Guangzhou Honda Guangzhou

Shenzhen Thunder Shenzhen

Tencent Shenzhen