在mapreduce中做分布式缓存的问题

一、问题描述：

主要解决一个问题，就是两个表做join，两个表都够大，单个表都无法装入内存。

怎么做呢？思路就是对做join的字段做排序两个表都排序，然后针对一个表a逐行读取，希望能够在内存中加载到另一个表b的数据，针对表a当前记录希望b的对应记录在内存中，这就是缓存的作用，希望命中率越高越好！

这个问题其实关键就是做缓存！

如下的情形是针对两个表做join的字段是两个，比如字段1字段2，做法是将表b 按照字段1分成多个文件，然后每个文件内按照字段2做好排序；表a也是同样的操作！

表b做晚切割后会有多个文件，需要做个元数据结构就是将这个partition内的字段2的最小值和最大值以及partition的名字这个3元组存储起来，作用就是当表a中一个记录来了后，能够知道对应的partition文件的名字，--注意这个元数据结构是永久存在内存的数据结构是就是(BeginEndPosition）。

存储一个hashMap（CustomLRUMap），key是partition文件名字，value是这个partition的数据，上一部得到了partition名字，就去这个map坐查询，如果没有就去加载。

二、具体描述

就是针对每次mapreduce的计算的时候希望通过一个缓存可以做做些查找，希望针对map或者reduce到的每条记录可以直接在内存中找到数据，如果找不到那么需要加载到内存！

这个索引的结构也就是 <分区文件名字,开始position,结束position> 这个三元组。

 

原始数据如上图所示，现在还需要一个meta data去组织数据

 

比如固定key1以后的按照key2做排序后split形成的partition文件如下：

这个文件就是最后的partition文件，注意：

Note that each of these partitioned files has a range of position values (since they are sorted by position). We will use these ranges in our cache implementation. Therefore, given a chromosome_id=1 and a position, we know exactly which partition holds the result of a query. Let’s look at the content of one of these sorted partitioned files如下是partition文件的内容，最左边的就是position（就是key2）字段，然后这个partition name是和key1有关系的: 

三、代码组织

0、首先组织元素据结构

You can see that all positions are sorted within each partition. To support meta‐ data for all partitions using LRU Map, we need an additional data structure to keep track of (begin, end) positions. For each partitioned file we will keep the (partition name, begin, end) information. 

伪代码：

>>BeginEndPosition对象实现了 the partition data structure such that you can get the database name for a given composite key.--作用就是根据chrId+position得到database name。
>> 注意MapDBEntry class 代表了 sorted partition of 64MB as a Map data structure implemented in MapDB.
the MapDBEntry class defines a single entry of a MapDB object ,比如new一个MapDBEntry对象的过程

public static MapDBEntry create(String dbName){
DB db=DBMaker.newFileDB(new File(dbName)).closeOnJvmShutDown().readOnly().make();
Map<String,String> map=db.getTreeMap("collectionName");
//可以从外村加载数据到map中去
MapDBEntry entry=new MapDBEntry(db,map);
return entry;
}

>>1、然后是cacheManager的初始化过程分析，最为关键的数据结构就是 theCustomLRUMap，这玩意的key是 dbname，value就是一个partition数据，其实就是文件名和数据的对应关系

注意cacheManage管理的是每一个partition，所以做替换内存操作的是每一个partition的操作！！！！,这个初始化过程放在setup()方法内，因为setup（）执行时间是

Setup gets called exactly once for each mapper, before map() gets called the first time.

It's a good place to do configuration or setup that can be shared across many calls to map

public static void init() throws Exception{
if(initialized)
return;
//注意这里的map类型 value是一个MapDBEntry类型的，其实
//这个数据结构说白了就是map中套map的类型
theCustomLRUMap=new CustomLRUMap<String,MapDBEntry<String,String>>(theLRUMapSize);
 7 beginEnd=new BeginEndPosition(mapdbBeginEndDirName);//元数据加载过程
 8 beginEnd.build(mapdbRootDirName);
initilized=true;

}

>>2、然后是使用
//首先是getDBName()

public static String getDBName(String key1,String key2){
List<Interval> results=beginEnd.query(key1,key2);
if(results==null || results.isEnpty()||results.size()==0) return null;
else return results.get(0).db();
}
//
public static String get(String key1,String key2) throws Exception{
String dbName=getDBName(key1,key2);
if(dbName==null) return null;
MapDBEntry<String,String> entry=theCustomLRUMap.get(dbName);
if(entry==null){
//需要做替换了
entry=MapDBEntryFactory.create(dbName);
theCustomLRUMap.put(dbName,entry);
}
return entry.getValue(key2);
}