简约的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。

希望大家都能在自己的机器上跑成功。源码都在上面了我就不放链接了。

原文发布于微信公众号 - 一名叫大蕉的程序员(DaBananaTalk)

原文发表时间:2017-07-02

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Jimoer

Java设计模式学习记录-桥接模式

这次介绍结构型设计模式中的第二种模式,桥接模式。 使用桥接模式的目的就是为了解耦,松散的耦合更利于扩展,但是会增加相应的代码量和设计难度。

722
来自专栏后端技术探索

php进阶

基本数据类型和数组都为真复制,即为真副本,当属性为对象时,为假复制,改变副本仍会影响原对象.解决方案:

1561
来自专栏挖掘大数据

Generator:化异步为同步

一、Promise并非完美 我在上一话中介绍了Promise,这种模式增强了事件订阅机制,很好地解决了控制反转带来的信任问题、硬编码回调执行顺序造成的“回调金...

2147
来自专栏吉浦迅科技

DAY37:阅读不同存储器的修饰符

1274
来自专栏游戏开发那些事

【Unity游戏开发】Lua中的os.date和os.time函数

  最近马三在工作中经常使用到了lua 中的 os.date( ) 和 os.time( )函数,不过使用的时候都是不得其解,一般都是看项目里面怎么用,然后我就...

1684
来自专栏腾讯IVWEB团队的专栏

响应式编程中 Stream 对象的实现原理

这篇文章介绍一种编程泛型,叫做响应式编程。将响应式称作“编程泛型”可能有些夸大其作用范畴,不过通过引入响应式确实会改变我们对特定问题的思考方法,就像刚接触red...

5380
来自专栏跟着阿笨一起玩NET

采用左右值编码来存储无限分级树形结构的数据库表设计

该设计方案的优点是:只用一条查询语句即可得到某个根节点及其所有子孙节点的先序遍历。由于消除了递归,在数据记录量较大时,可以大大提高列表效率。但是,这种编码方案由...

3721
来自专栏用户2442861的专栏

教你如何迅速秒杀掉:99%的海量数据处理面试题

   一般而言,标题含有“秒杀”,“99%”,“史上最全/最强”等词汇的往往都脱不了哗众取宠之嫌,但进一步来讲,如果读者读罢此文,却无任何收获,那么,我也甘愿...

1452
来自专栏数据魔术师

运筹学教学|分支定界法解带时间窗的车辆路径规划问题(附代码及详细注释)

67510
来自专栏听雨堂

数据库范式

关系数据库设计之时是要遵守一定的规则的。尤其是数据库设计范式 现简单介绍1NF(第一范式),2NF(第二范式),3NF(第三范式)和BCNF,另有第四范式和第五...

1976

扫码关注云+社区

领取腾讯云代金券