简约的JAVA版本MapReduce和日常No.25
简约的JAVA版本MapReduce和日常No.25
昨天做了一个小调查，说看看想看些啥。大概的分布是这样的，一个1代表一个投票。看来还是2、3比较多。
11111 希望看到"算法"回复1。 111111111111 希望看到"技术细节"回复2。111111111 希望看到"成长和读书"分享回复3。还好多人说想看我长啥样，嘛，在我比较正经的时候，就长下面这样。
大图预警！！！！
日常呢，就长这样。
长这样。
好了切入正题，今天开始挖一个新坑，就是实现一些基于MapReduce的一些图算法，比如Pregel啊，PageRank啊，LPA啊，SLPA啊等等，坑很大，非常大，慢慢写吧，都不会讲非常难的理论问题，以代码细节为主。。
先上一个我思维拓展的时候写得java实现的MapReduce的基础版本吧，写得不是很好，我也在慢慢完善，Go语言版本的还在写，真是惭愧感觉一直在吃老本。
今天实现的一个内容是，将一个List<Integer>进行map操作变成另外一个List，然后通过reduce进行加和。
灵感来源来自于《MapReduce: Simplified Data Processing on Large Clusters 》这篇论文，大家可以看看我之前的文章，在了解完什么是Mapreduce。然后先去看看这篇论文，启发很多。
首先我们从两个接口入手，MapFunction和ReduceFunction，这是MapReduce的两个灵魂接口，由使用者去定义，这里我定义的都是最最简单的版本，暂时并没有进行泛化的能力。
MapFunction定义了一个接口，类型为V，然后通过一个叫map的方法，输出一个类型为V的值。
public interface MapFunction<V> { V map(V target); }ReduceFunction定义了一个接口，类型为V，然后通过一个叫reduce的方法，通过聚合两个V类型的值，输出一个类型为V的值。
public interface ReduceFunction<V> { V reduce(V A,V B); }上面两个方法定义了MapReduce的核心内容，就是任务切分和任务聚合。有小伙伴不理解这里为什么使用泛型，因为作为一个框架来说，我是不知道使用者想使用什么样的类型进行计算的（虽然这里我知道我接下来就要用Integer进行计算了），所以必须不能指定类型，否则这个框架就永远只能用Integer类型了。
那我们的map和reduce任务要跑在哪里呢？有小伙伴说跑在分布式环境里。对没错，最终目的是跑在分布式环境里。但是在这里，咱就偷个懒，先用多线程来模拟这个过程，并且使用内存来作为消息机制。
我是i5双核的CPU，经验值下面，只有两个cpu的话，创建4个线程对于性能来说比单线程好。（毕竟线程切换存在开销，控制得不好多线程肯定是比单线程慢的，不服来辩）
public class CPUs {
public static final int threads = 4;
private static final java.util.concurrent.ExecutorService pool = Executors.newFixedThreadPool(threads);
public static Future submit(Callable task){
return pool.submit(task);
}
public static void execute(Runnable task){
pool.execute(task);
}
public static void shutdown(){
pool.shutdown();
}
}好了，MapFunction有了，CPUs也有了，接下来可以开始写提交器了。任务提交器是什么东西呢，就是把一个map任务进行切分，并且交给多个线程去异步执行，然后最终把结果汇总还给客户端的一个类。下面的类都比较大，建议在电脑端看。
这个类做了什么事呢？就是把List封装起来，然后把任务分发给多个线程去执行，使用CountDownLatch来保证所有的线程都已经完成计算，然后再把结果返回给客户端。
public class MapSubmitter<V> {
private List<V> target ;
private int length;
public MapSubmitter(List<V> target){
this.target = target;
this.length = target.size();
}
public List<V> map(final MapFunction<V> mapFunction){
final CountDownLatch countDownLatch = new CountDownLatch(length);
final List<V> result = new ArrayList<V>();
for(int i = 0 ; i < length ; i++) {
final V current = target.get(i);
final int currentIndex = i;
try {
Future<V> future = CPUs.submit(new Callable<V>() {
public V call() throws Exception {
V result = mapFunction.map(current);
//Printer.println(currentIndex);
return result;
}
});
result.add(i,future.get());
countDownLatch.countDown();
}catch (InterruptedException e) {
e.printStackTrace();
}catch (ExecutionException e) {
e.printStackTrace();
}
}
try{
countDownLatch.await();
} catch (InterruptedException e) { }
finally {
return result;
}
}
}这个类又做了什么事呢？List封装起来，交给很多线程去执行，然后维护一个最终的结果类V，并为这个结果提供线程安全的保护，避免因为多线程操作同一个结果造成结果错误。
public class ReduceSubmitter<V> {
private List<V> target ;
private int length;
private V result ;
Lock lock = new ReentrantLock();
public ReduceSubmitter(List<V> target){
this.target = target;
this.length = target.size();
this.result = target.get(0);
}
public V reduce(final ReduceFunction<V> reduceFunction){
final CountDownLatch countDownLatch = new CountDownLatch(length);
countDownLatch.countDown();
for(int i = 1 ; i < length ; i ++) {
final V current = target.get(i);
CPUs.execute(new Runnable() {
public void run() {
lock.lock();
V next = reduceFunction.reduce(ReduceSubmitter.this.result,current);
ReduceSubmitter.this.result = next;
lock.unlock();
countDownLatch.countDown();
}
});
}
try{
countDownLatch.await();
} catch (InterruptedException e) { }
finally {
return this.result;
}
}
}
好咯，写完了就开始测试了,主要就创建一个长度为10的数组，然后进行map操作把每一个值都进行平方，然后通过reduce操作进行求和，代码比较简单就不一一细说了，有啥问题后台留言交流。
public class TestMapReduce {
public static void main(String[] args){
//仅仅是为了耗时而模拟的一个好像很复杂的操作，不然太快了。
final int junkTime = 1000000;
//初始化一个想进行操作的数组
List<Integer> integerList = new ArrayList<Integer>();
for(int i = 0 ; i < 10 ; i++){
integerList.add(i);
}
int length = integerList.size();
// printer.printList(integerList);
Long start = System.currentTimeMillis();
//进行map操作并返回结果
MapSubmitter<Integer> mapSubmitter = new MapSubmitter<Integer>(integerList);
integerList = mapSubmitter.map(new MapFunction<Integer>() {
public Integer map(Integer target) {
Double b = 0D;
for(int i = 0 ; i <junkTime;i++){
b += Math.exp(i);
}
return target * target;
}
});
Printer.println("mapreduce cost time:" + (System.currentTimeMillis() - start));
start = System.currentTimeMillis();
//进行reduce操作并返回结果
ReduceSubmitter<Integer> reduceSubmitter = new ReduceSubmitter<Integer>(integerList);
Integer resultInteger = reduceSubmitter.reduce(new ReduceFunction<Integer>() {
public Integer reduce(Integer A, Integer B) {
Double b = 0D;
for(int i = 0 ; i <junkTime;i++){
b += Math.exp(i);
}
return A+B;
}
});
Printer.println("reduce cost time:" + (System.currentTimeMillis() - start));
CPUs.shutdown(); }
}好啦，今天的MapReduce就说到这里。经过我的实验，无论多少次实验，都是比单线程快那么一丢丢的，这都要得益于那个耗时的操作，模糊了线程切换带来的时间损耗，毕竟不怎么耗时的操作来说，单线程其实是绝对比多线程快的。
细心的同学会发现，好像这个并不符合论文里面的标准吖。嗯呐是的，这个只是我心血来潮写的简单版本。问题有诸如，我们上面的map操作好像不能变成其他类型吖，怎么实现WordCount呢？以及Driver好像没有进行任务切分和分发吖？好像也没有suffle操作啊？好像整个过程也不是严格多线程的吖，怎么办呢？下一次给大家分享一个更加完整的MapReduce。
希望大家都能在自己的机器上跑成功。源码都在上面了我就不放链接了。