博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
map侧连接
阅读量:7117 次
发布时间:2019-06-28

本文共 3631 字,大约阅读时间需要 12 分钟。

两个数据集中一个非常小,可以让小数据集存入缓存。在作业开始这些文件会被复制到运行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

转载于:https://www.cnblogs.com/luminous1/p/8386916.html

你可能感兴趣的文章
做程序员有什么好处
查看>>
Android报unable to instantiate application怎么解决?
查看>>
maven远程发布项目
查看>>
centos 初始化静态 ip
查看>>
Java匿名内部类
查看>>
expdp 和impdp使用之一(不同用户和不用表空间)
查看>>
APP測試
查看>>
4-11高项作业
查看>>
Python——字符串1
查看>>
BZOJ 1227 虔诚的墓主人
查看>>
[转载]在线考试javaScript倒计时
查看>>
嵌入式 Linux进程间通信(一)——进程
查看>>
Java项目命名规范
查看>>
Python多重装饰器
查看>>
查看raid信息
查看>>
Spring学习总结(4)——Spring AOP教程
查看>>
jsp连接mysql和mssql的简单总结
查看>>
RabbitMQ学习总结(4)——分发任务在多个工作者之间实例教程
查看>>
RabbitMQ学习总结(7)——Spring整合RabbitMQ实例
查看>>
[cocos2d-x 3.2]学习绘图API
查看>>