MapReduce原理与设计思想（转载：http://blog.jobbole.com/80619/）-低调大师

MapReduce原理与设计思想（转载：http://blog.jobbole.com/80619/）

2017-06-01 517

简单解释 MapReduce 算法

一个有趣的例子

你想数出一摞牌中有多少张黑桃。直观方式是一张一张检查并且数出有多少张是黑桃

MapReduce方法则是

给在座的所有玩家中分配这摞牌
让每个玩家数自己手中的牌有几张是黑桃然后把这个数目汇报给你
你把所有玩家告诉你的数字加起来得到最后的结论

拆分

MapReduce合并了两种经典函数

映射Mapping对集合里的每个目标应用同一个操作。即如果你想把表单里每个单元格乘以二那么把这个函数单独地应用在每个单元格上的操作就属于mapping。
化简Reducing 遍历集合中的元素来返回一个综合的结果。即输出表单里一列数字的和这个任务属于reducing。

重新审视上面的例子

重新审视我们原来那个分散纸牌的例子我们有MapReduce数据分析的基本方法。友情提示这不是个严谨的例子。在这个例子里人代表计算机因为他们同时工作所以他们是个集群。在大多数实际应用中我们假设数据已经在每台计算机上了 – 也就是说把牌分发出去并不是MapReduce的一步。事实上在计算机集群中如何存储文件是Hadoop的真正核心。

通过把牌分给多个玩家并且让他们各自数数你就在并行执行运算因为每个玩家都在同时计数。这同时把这项工作变成了分布式的因为多个不同的人在解决同一个问题的过程中并不需要知道他们的邻居在干什么。

通过告诉每个人去数数你对一项检查每张牌的任务进行了映射。你不会让他们把黑桃牌递给你而是让他们把你想要的东西化简为一个数字。

另外一个有意思的情况是牌分配得有多均匀。MapReduce假设数据是洗过的shuffled- 如果所有黑桃都分到了一个人手上那他数牌的过程可能比其他人要慢很多。

如果有足够的人的话问一些更有趣的问题就相当简单了 – 比如“一摞牌的平均值二十一点算法是什么”。你可以通过合并“所有牌的值的和是什么”及“我们有多少张牌”这两个问题来得到答案。用这个和除以牌的张数就得到了平均值。

MapReduce算法的机制要远比这复杂得多但是主体思想是一致的 – 通过分散计算来分析大量数据。无论是Facebook、NASA还是小创业公司MapReduce都是目前分析互联网级别数据的主流方法。

Hadoop中的MapReduce

大规模数据处理时MapReduce在三个层面上的基本构思

如何对付大数据处理分而治之

对相互间不具有计算依赖关系的大数据实现并行最自然的办法就是采取分而治之的策略

上升到抽象模型Mapper与Reducer

MPI等并行计算方法缺少高层并行编程模型为了克服这一缺陷MapReduce借鉴了Lisp函数式语言中的思想用Map和Reduce两个函数提供了高层的并行编程抽象模型

上升到构架统一构架为程序员隐藏系统层细节

MPI等并行计算方法缺少统一的计算框架支持程序员需要考虑数据存储、划分、分发、结果收集、错误恢复等诸多细节为此MapReduce设计并提供了统一的计算框架为程序员隐藏了绝大多数系统层面的处理细节

1.对付大数据处理-分而治之

什么样的计算任务可进行并行化计算

并行计算的第一个重要问题是如何划分计算任务或者计算数据以便对划分的子任务或数据块同时进行计算。但一些计算问题恰恰无法进行这样的划分

Nine women cannot have a baby in one month!

例如Fibonacci函数: Fk+2 = Fk + Fk+1

前后数据项之间存在很强的依赖关系只能串行计算

结论不可分拆的计算任务或相互间有依赖关系的数据无法进行并行计算

大数据的并行化计算

一个大数据若可以分为具有同样计算过程的数据块并且这些数据块之间不存在数据依赖关系则提高处理速度的最好办法就是并行计算

例如假设有一个巨大的2维数据需要处理(比如求每个元素的开立方)其中对每个元素的处理是相同的,并且数据元素间不存在数据依赖关系,可以考虑不同的划分方法将其划分为子数组,由一组处理器并行处理

2.构建抽象模型-Map和Reduce

借鉴函数式设计语言Lisp的设计思想

函数式程序设计(functional programming)语言Lisp是一种列表处理语言(List processing)是一种应用于人工智能处理的符号式语言由MIT的人工智能专家、图灵奖获得者John McCarthy于1958年设计发明。

Lisp定义了可对列表元素进行整体处理的各种操作如

如(add #(1 2 3 4) #(4 3 2 1)) 将产生结果 #(5 5 5 5)

Lisp中也提供了类似于Map和Reduce的操作

如: (map ‘vector #+ #(1 2 3 4 5) #(10 11 12 13 14))

通过定义加法map运算将2个向量相加产生结果#(11 13 15 17 19)

(reduce #’+ #(11 13 15 17 19)) 通过加法归并产生累加结果75

Map: 对一组数据元素进行某种重复式的处理

Reduce: 对Map的中间结果进行某种进一步的结果整

关键思想为大数据处理过程中的两个主要处理操作提供一种抽象机制

MapReduce中的Map和Reduce操作的抽象描述

MapReduce借鉴了函数式程序设计语言Lisp中的思想定义了如下的Map和Reduce两个抽象的编程接口由用户去编程实现:

map: (k1; v1) → [(k2; v2)]

输入键值对(k1; v1)表示的数据

处理文档数据记录(如文本文件中的行或数据表格中的行)将以“键值对”形式传入map函数map函数将处理这些键值对并以另一种键值对形式输出处理的一组键值对中间结果　　　[(k2; v2)]

输出键值对[(k2; v2)]表示的一组中间数据

reduce: (k2; [v2]) → [(k3; v3)]

输入由map输出的一组键值对[(k2; v2)] 将被进行合并处理将同样主键下的不同数值合并到一个列表[v2]中故reduce的输入为(k2; [v2])

处理对传入的中间结果列表数据进行某种整理或进一步的处理,并产生最终的某种形式的结果输出[(k3; v3)] 。

输出最终输出结果[(k3; v3)]

Map和Reduce为程序员提供了一个清晰的操作接口抽象描述

各个map函数对所划分的数据并行处理从不同的输入数据产生不同的中间结果输出

各个reduce也各自并行计算各自负责处理不同的中间结果数据集合进行reduce处理之前,必须等到所有的map函数做完因此,在进入reduce前需要有一个同步障(barrier);这个阶段也负责对map的中间结果数据进行收集整理(aggregation & shuffle)处理,以便reduce更有效地计算最终结果最终汇总所有reduce的输出结果即可获得最终结果

基于MapReduce的处理过程示例—文档词频统计WordCount

设有4组原始文本数据

Text 1: the weather is good Text 2: today is good

Text 3: good weather is good Text 4: today has good weather

传统的串行处理方式(Java)

Java

  
    
     
        1
       

        2
       

        3
       

        4
       

        5
       

        6
       

        7
       

        8
       

        9
       

        10
       

        11
       

        12
       

        13
       

        14
       

        15
       

        16
       

        17
       
 
      
        String
        [
        ]
         
        text
         
        =
         
        new
         
        String
        [
        ]
         
        {
         “
        hello 
        world”
        ,
         “
        hello 
        every 
        one”
        ,
         “
        say 
        hello 
        to
         
        everyone 
        in
         
        the 
        world”
         
        ; 
       
 
        HashTable 
        ht
         
        =
         
        new
         
        HashTable
        (
        )
        ;
             
       
 
        for
        (
        i
         
        =
         
        0
        ;
         
        i
         
        <
         
        3
        ;
         
        ++
        i
        )
         
        { 
       
 
            
        StringTokenizer 
        st
         
        =
         
        new
         
        StringTokenizer
        (
        text
        [
        i
        ]
        )
        ;
          
       
 
            
        while
         
        (
        st
        .
        hasMoreTokens
        (
        )
        )
         
        {
           
       
 
                
        String
         
        word
         
        =
         
        st
        .
        nextToken
        (
        )
        ; 
       
 
                
        if
        (
        !
        ht
        .
        containsKey
        (
        word
        )
        )
         
        {
           
       
 
                    
        ht
        .
        put
        (
        word
        ,
         
        new
         
        Integer
        (
        1
        )
        )
        ; 
       
 
                
        }
         
        else
         
        { 
       
 
                    
        int
         
        wc
         
        =
         
        (
        (
        Integer
        )
        ht
        .
        get
        (
        word
        )
        )
        .
        intValue
        (
        )
         
        +
        1
        ;
        // 计数加1 
       
 
                    
        ht
        .
        put
        (
        word
        ,
         
        new
         
        Integer
        (
        wc
        )
        )
        ; 
       
 
                
        } 
       
 
            
        } 
       

        }
       
 
        for
         
        (
        Iterator 
        itr
        =
        ht
        .
        KeySet
        (
        )
        .
        iterator
        (
        )
        ;
          
        itr
        .
        hasNext
        (
        )
        ;
         
        )
         
        { 
       
 
            
        String
         
        word
         
        =
         
        (
        String
        )
        itr
        .
        next
        (
        )
        ;
          
       
 
            
        System
        .
        out
        .
        print
        (
        word
        +
         “
        :
         ”
        +
         
        (
        Integer
        )
        ht
        .
        get
        (
        word
        )
        +“
        ;
           ”
        )
        ; 
       
 
    

  
 

输出good: 5; has: 1; is: 3; the: 1; today: 2; weather: 3

基于MapReduce的处理过程示例—文档词频统计WordCount

MapReduce处理方式

使用4个map节点

map节点1:

输入(text1, “the weather is good”)

输出(the, 1), (weather, 1), (is, 1), (good, 1)

map节点2:

输入(text2, “today is good”)

输出(today, 1), (is, 1), (good, 1)

map节点3:

输入(text3, “good weather is good”)

输出(good, 1), (weather, 1), (is, 1), (good, 1)

map节点4:

输入(text3, “today has good weather”)

输出(today, 1), (has, 1), (good, 1), (weather, 1)

使用3个reduce节点

MapReduce处理方式

MapReduce伪代码(实现Map和Reduce两个函数)

C

        1
       
        2
       
        3
       
        4
       
        5
       
        6
       
        7
       
        8
       
        9
       
        10
       
        11
       
        12
       
        13
       
        Class
         
        Mapper 
        method 
        map
        (
        String
         
        input_key
        ,
         
        String
         
        input_value
        )
        : 
       
        // input_key: text document name  
       
        // input_value: document contents  
       
        for
         
        each
         
        word
         
        w
         
        in
         
        input_value
        :
          
        EmitIntermediate
        (
        w
        ,
         
        "1"
        )
        ;
          
        Class
         
        Reducer 
        method 
        reduce
        (
        String
         
        output_key
        ,
         
        Iterator 
        intermediate_values
        )
        :
          
        // output_key: a word  
       
        // output_values: a list of counts  
       
        int
         
        result
         
        =
         
        0
        ;
          
        for
         
        each
         
        v
         
        in
         
        intermediate_values
        :
          
        result
         
        +=
         
        ParseInt
        (
        v
        )
        ; 
       
        Emit
        (
        output
        _key
         
        result
        )
        ;

3.上升到构架-自动并行化并隐藏低层细节

如何提供统一的计算框架

MapReduce提供一个统一的计算框架可完成

计算任务的划分和调度

数据的分布存储和划分

处理数据与计算任务的同步

结果数据的收集整理(sorting, combining, partitioning,…)

系统通信、负载平衡、计算性能优化处理

处理系统节点出错检测和失效恢复

MapReduce最大的亮点

通过抽象模型和计算框架把需要做什么(what need to do)与具体怎么做(how to do)分开了为程序员提供一个抽象和高层的编程接口和框架

程序员仅需要关心其应用层的具体计算问题仅需编写少量的处理应用本身计算问题的程序代码

如何具体完成这个并行计算任务所相关的诸多系统层细节被隐藏起来,交给计算框架去处理从分布代码的执行到大到数千小到单个节点集群的自动调度使用

MapReduce提供的主要功能

任务调度提交的一个计算作业(job)将被划分为很多个计算任务(tasks), 任务调度功能主要负责为这些划分后的计算任务分配和调度计算节点(map节点或reducer节点); 同时负责监控这些节点的执行状态, 并负责map节点执行的同步控制(barrier); 也负责进行一些计算性能优化处理, 如对最慢的计算任务采用多备份执行、选最快完成者作为结果

数据/代码互定位为了减少数据通信一个基本原则是本地化数据处理(locality)即一个计算节点尽可能处理其本地磁盘上所分布存储的数据这实现了代码向数据的迁移当无法进行这种本地化数据处理时再寻找其它可用节点并将数据从网络上传送给该节点(数据向代码迁移)但将尽可能从数据所在的本地机架上寻找可用节点以减少通信延迟

出错处理以低端商用服务器构成的大规模MapReduce计算集群中,节点硬件(主机、磁盘、内存等)出错和软件有bug是常态因此,MapReducer需要能检测并隔离出错节点并调度分配新的节点接管出错节点的计算任务

分布式数据存储与文件管理海量数据处理需要一个良好的分布数据存储和文件管理系统支撑,该文件系统能够把海量数据分布存储在各个节点的本地磁盘上,但保持整个数据在逻辑上成为一个完整的数据文件为了提供数据存储容错机制,该文件系统还要提供数据块的多备份存储管理能力

Combiner和Partitioner:为了减少数据通信开销,中间结果数据进入reduce节点前需要进行合并(combine)处理,把具有同样主键的数据合并到一起避免重复传送; 一个reducer节点所处理的数据可能会来自多个map节点, 因此, map节点输出的中间结果需使用一定的策略进行适当的划分(partitioner)处理保证相关数据发送到同一个reducer节点

基于Map和Reduce的并行计算模型

4.MapReduce的主要设计思想和特征

1、向“外”横向扩展而非向“上”纵向扩展Scale “out”, not “up”

即MapReduce集群的构筑选用价格便宜、易于扩展的大量低端商用服务器而非价格昂贵、不易扩展的高端服务器SMP低端服务器市场与高容量Desktop PC有重叠的市场因此由于相互间价格的竞争、可互换的部件、和规模经济效应使得低端服务器保持较低的价格基于TPC-C在2007年底的性能评估结果,一个低端服务器平台与高端的共享存储器结构的服务器平台相比,其性价比大约要高4倍;如果把外存价格除外,低端服务器性价比大约提高12倍对于大规模数据处理由于有大量数据存储需要显而易见基于低端服务器的集群远比基于高端服务器的集群优越这就是为什么MapReduce并行计算集群会基于低端服务器实现

2、失效被认为是常态Assume failures are common

MapReduce集群中使用大量的低端服务器(Google目前在全球共使用百万台以上的服务器节点),因此节点硬件失效和软件出错是常态因而一个良好设计、具有容错性的并行计算系统不能因为节点失效而影响计算服务的质量任何节点失效都不应当导致结果的不一致或不确定性任何一个节点失效时其它节点要能够无缝接管失效节点的计算任务当失效节点恢复后应能自动无缝加入集群而不需要管理员人工进行系统配置MapReduce并行计算软件框架使用了多种有效的机制如节点自动重启技术使集群和计算框架具有对付节点失效的健壮性能有效处理失效节点的检测和恢复。

3、把处理向数据迁移Moving processing to the data

　　传统高性能计算系统通常有很多处理器节点与一些外存储器节点相连如用区域存储网络(SAN,Storage Area Network)连接的磁盘阵列因此大规模数据处理时外存文件数据I/O访问会成为一个制约系统性能的瓶颈。为了减少大规模数据并行计算系统中的数据通信开销代之以把数据传送到处理节点(数据向处理器或代码迁移)应当考虑将处理向数据靠拢和迁移。MapReduce采用了数据/代码互定位的技术方法计算节点将首先将尽量负责计算其本地存储的数据,以发挥数据本地化特点(locality),仅当节点无法处理本地数据时再采用就近原则寻找其它可用计算节点并把数据传送到该可用计算节点。

4、顺序处理数据、避免随机访问数据Process data sequentially and avoid random access

　大规模数据处理的特点决定了大量的数据记录不可能存放在内存、而只可能放在外存中进行处理。磁盘的顺序访问和随即访问在性能上有巨大的差异

例100亿(1010)个数据记录(每记录100B,共计1TB)的数据库

更新1%的记录(一定是随机访问)需要1个月时间而顺序访问并重写所有数据记录仅需1天时间

　MapReduce设计为面向大数据集批处理的并行计算系统所有计算都被组织成很长的流式操作以便能利用分布在集群中大量节点上磁盘集合的高传输带宽。

5、为应用开发者隐藏系统层细节Hide system-level details from the application developer

　　软件工程实践指南中专业程序员认为之所以写程序困难是因为程序员需要记住太多的编程细节(从变量名到复杂算法的边界情况处理)这对大脑记忆是一个巨大的认知负担,需要高度集中注意力而并行程序编写有更多困难如需要考虑多线程中诸如同步等复杂繁琐的细节由于并发执行中的不可预测性程序的调试查错也十分困难大规模数据处理时程序员需要考虑诸如数据分布存储管理、数据分发、数据通信和同步、计算结果收集等诸多细节问题MapReduce提供了一种抽象机制将程序员与系统层细节隔离开来程序员仅需描述需要计算什么(what to compute), 而具体怎么去做(how to compute)就交由系统的执行框架处理这样程序员可从系统层细节中解放出来而致力于其应用本身计算问题的算法设计

6、平滑无缝的可扩展性Seamless scalability

主要包括两层意义上的扩展性数据扩展和系统规模扩展。理想的软件算法应当能随着数据规模的扩大而表现出持续的有效性性能上的下降程度应与数据规模扩大的倍数相当在集群规模上要求算法的计算性能应能随着节点数的增加保持接近线性程度的增长绝大多数现有的单机算法都达不到以上理想的要求把中间结果数据维护在内存中的单机算法在大规模数据处理时很快失效从单机到基于大规模集群的并行计算从根本上需要完全不同的算法设计奇妙的是MapReduce几乎能实现以上理想的扩展性特征。多项研究发现基于MapReduce的计算性能可随节点数目增长保持近似于线性的增长

微信关注我们

原文链接：https://yq.aliyun.com/articles/100925

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

apache开源项目--HydraBase

Facebook 在官方博客上宣布推出HBase数据库的升级版——HydraBase， Facebook是HBase的重度用户，Facebook的HBase数据库系统存储着Facebook的很多关键业务数据，包括内部监控系统、搜索索引、流数据分析以及数据抓取等。HydraBase相比HBase稳定性和可用性更高，可以减少服务器宕机时间。在HBase系统中，数据分片存储于很多区域，如果某个区域服务器宕机，其域内数据都需要迁移到另外一个域服务器。Facebook指出，虽然HBase能够自动恢复，但是恢复时间过长。目前该项目已经捐赠给 Apache 基金会。 HydraBase的典型部署模型 HydraBase能够让一个数据域分布在多个域服务器中，域服务器之间能相互备份，因此能够大大减少数据恢复所用的时间。Facebook声称HydraBase能将Facebook全年的宕机时间缩减到不到5分钟。 Facebook目前正在测试HydraBase，并计划在生产集群中逐步开始部署。发布示例：单节点故障：数据中心故障： Facebook声称HydraBase能将Facebook全年的宕机...

2017-06-01

548

工程结构：在整个案例过程中，代码如下： WordCountMapper的代码如下： package cn.toto.bigdata.mr.wc; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; /** * 这里的Mapper是hadoop-mapreduce-client-core-2.8.0.jar中的内容 * Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> * KEYIN :是指框架读取到的数据的key的类型，在默认的InputFormat下，读到的key是一行文本的起始偏移量，所以key的类型是Long * VALUEIN :是指框架读取到的数据的value的类型，在默认的InputFormat下，读到的...

2017-06-02

659

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。