mapreduce 订阅
MapReduce是一种编程模型,用于大规模数据集(大于1TB)的并行运算。概念"Map(映射)"和"Reduce(归约)",是它们的主要思想,都是从函数式编程语言里借来的,还有从矢量编程语言里借来的特性。它极大地方便了编程人员在不会分布式并行编程的情况下,将自己的程序运行在分布式系统上。 当前的软件实现是指定一个Map(映射)函数,用来把一组键值对映射成一组新的键值对,指定并发的Reduce(归约)函数,用来保证所有映射的键值对中的每一个共享相同的键组。 展开全文
MapReduce是一种编程模型,用于大规模数据集(大于1TB)的并行运算。概念"Map(映射)"和"Reduce(归约)",是它们的主要思想,都是从函数式编程语言里借来的,还有从矢量编程语言里借来的特性。它极大地方便了编程人员在不会分布式并行编程的情况下,将自己的程序运行在分布式系统上。 当前的软件实现是指定一个Map(映射)函数,用来把一组键值对映射成一组新的键值对,指定并发的Reduce(归约)函数,用来保证所有映射的键值对中的每一个共享相同的键组。
信息
思想来源
Google的几篇论文
本    质
一种编程模型
特    点
分布可靠
用    途
大规模数据集的并行运算
应    用
大规模的算法图形处理、文字处理
外文名
MapReduce
MapReduce定义
MapReduce是面向大数据并行处理的计算模型、框架和平台,它隐含了以下三层含义:1)MapReduce是一个基于集群的高性能并行计算平台(Cluster Infrastructure)。它允许用市场上普通的商用服务器构成一个包含数十、数百至数千个节点的分布和并行计算集群。2)MapReduce是一个并行计算与运行软件框架(Software Framework)。它提供了一个庞大但设计精良的并行计算软件框架,能自动完成计算任务的并行化处理,自动划分计算数据和计算任务,在集群节点上自动分配和执行任务以及收集计算结果,将数据分布存储、数据通信、容错处理等并行计算涉及到的很多系统底层的复杂细节交由系统负责处理,大大减少了软件开发人员的负担。3)MapReduce是一个并行程序设计模型与方法(Programming Model & Methodology)。它借助于函数式程序设计语言Lisp的设计思想,提供了一种简便的并行程序设计方法,用Map和Reduce两个函数编程实现基本的并行计算任务,提供了抽象的操作和并行编程接口,以简单方便地完成大规模数据的编程和计算处理 [1]  。
收起全文
精华内容
参与话题
问答
  • MapReduce

    千次阅读 多人点赞 2019-06-14 15:33:14
    一、MapReduce简介 什么是MapReduceMapreduce 是一个分布式运算程序的编程框架,是用户开发“基于 hadoop 的数据分析 应用”的核心框架。 Mapreduce 核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一...

    一、MapReduce简介

    • 什么是MapReduce?
      Mapreduce 是一个分布式运算程序的编程框架,是用户开发“基于 hadoop 的数据分析 应用”的核心框架。
      Mapreduce 核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的 分布式运算程序,并发运行在一个hadoop 集群上

    • Hadoop的四个组件
      HDFS:分布式存储系统
      MapReduce:分布式计算系统
      YARN: hadoop 的资源调度系统
      Common: 以上三大组件的底层支撑组件,主要提供基础工具包和 RPC 框架等

    • 为什么需要 MapReduce?
      (1) 海量数据在单机上处理因为硬件资源限制,无法胜任
      (2) 而一旦将单机版程序扩展到集群来分布式运行,将极大增加程序的复杂度和开发难度
      (3) 引入 MapReduce 框架后,开发人员可以将绝大部分工作集中在业务逻辑的开发上,而将 分布式计算中的复杂性交由框架来处理
      (4)从而提高开发效率,减轻开发者的负担

    • MapReduce的优点
      1.易于编程
      2.良好的扩展性
      3.高容错性
      4.适合PB级别以上的大数据的分布式离线批处理

    • MapReduce的缺点
      1.难以实时计算(MapReduce处理的是存储在本地磁盘上的离线数据)
      2.不能流式计算(MapReduce设计处理的数据源是静态的)
      3.难以DAG计算MapReduce这些并行计算大都是基于非循环的数据流模型,也就是说,一次计算过程中,不同计算节点之间保持高度并行,这样的数据流模型使得那些需要反复使用一个特定数据集的迭代算法无法高效地运行

    二、MapReduce的工作流程

    • input
      MapReduce需要把要执行的大文件数据进行切割(split)
      每一个输入分片(input split)对应一个map任务
      输入分片(input split)存储的并非数据本身而是一个分片长度和一个记录数据位置的数组
      输入分片(input split)和HDFS(hadoop2.0之后)的block(块)关系很密切,HDFS(hadoop2.0之后)的block块的大小默认是128M,如果我们执行的大文件是128x10M,MapReduce会分为10个map任务,每个map任务都存在于它所要计算的block(块)的DataNode上
    • map
      程序员编写的map函数
      因此map函数效率相对好控制
      而且一般map操作都是本地化操作也就是在数据存储节点上进行
    • shuffle
      负责将map生成的数据传递给reduce
      因此shuffle分为在map的执行过程和在reduce的执行过程
    • reduce
      负责shuffle传递过来的数据进行合并
    • output
      最后负责输出reduce中的数据信息
      在这里插入图片描述
    展开全文
  • MapReduce的通俗理解与入门

    万次阅读 多人点赞 2018-06-07 11:10:20
    看这篇文章请出去跑两圈,然后泡一壶茶,边喝茶,边看,看完你就对hadoop 与MapReduce的整体有所了解了。【前言】Hadoop是一个实现了Google云计算系统的开源系统,包括并行计算模型Map/Reduce、分布式文件系统HDFS,...

    看这篇文章请出去跑两圈,然后泡一壶茶,边喝茶,边看,看完你就对hadoop 与MapReduce的整体有所了解了。

    【前言】

    Hadoop是一个实现了Google云计算系统的开源系统,包括并行计算模型Map/Reduce、分布式文件系统HDFS,以及分布式数据库Hbase,同时Hadoop的相关项目也很丰富,包括ZooKeeper,Pig,Chukwa,Hive,Hbase,Mahout,flume等。

    本文就hadoop的并行分布式计算模型MapReduce做一个简单的入门介绍。

    【什么是Map/Reduce】

    看看下面的各种解释:

    (1)MapReduce是hadoop的核心组件之一,hadoop要实现分布式需要包括两部分,一部分是分布式文件系统hdfs,一部分是分布式计算框架mapreduce,缺一不可,也就是说,可以通过mapreduce很容易在hadoop平台上进行分布式的计算编程。
    (2)Mapreduce是一种编程模型,是一种方法,抽象理论。
    (3)下面是一个关于一个程序员是如何跟妻子讲解什么是MapReduce,文章很长请耐心的看。

    我问妻子:“你真的想要弄懂什么是MapReduce?” 她很坚定的回答说“是的”。 

    因此我问道:

    我: 你是如何准备洋葱辣椒酱的?(以下并非准确食谱,请勿在家尝试)
    妻子: 我会取一个洋葱,把它切碎,然后拌入盐和水,最后放进混合研磨机里研磨。这样就能得到洋葱辣椒酱了。但这和MapReduce有什么关系?
    我: 你等一下。让我来编一个完整的情节,这样你肯定可以在15分钟内弄懂MapReduce。
    妻子: 好吧。
    我:现在,假设你想用薄荷、洋葱、番茄、辣椒、大蒜弄一瓶混合辣椒酱。你会怎么做呢?
    妻子: 我会取薄荷叶一撮,洋葱一个,番茄一个,辣椒一根,大蒜一根,切碎后加入适量的盐和水,再放入混合研磨机里研磨,这样你就可以得到一瓶混合辣椒酱了。

    我: 没错,让我们把MapReduce的概念应用到食谱上。Map和Reduce其实是两种操作,我来给你详细讲解下。Map(映射) 把洋葱、番茄、辣椒和大蒜切碎,是各自作用在这些物体上的一个Map操作。所以你给Map一个洋葱,Map就会把洋葱切碎。 同样的,你把辣椒,大蒜和番茄一一地拿给Map,你也会得到各种碎块。 所以,当你在切像洋葱这样的蔬菜时,你执行就是一个Map操作。 Map操作适用于每一种蔬菜,它会相应地生产出一种或多种碎块,在我们的例子中生产的是蔬菜块。在Map操作中可能会出现有个洋葱坏掉了的情况,你只要把坏洋葱丢了就行了。所以,如果出现坏洋葱了,Map操作就会过滤掉坏洋葱而不会生产出任何的坏洋葱块。Reduce(化简)这一阶段,你将各种蔬菜碎都放入研磨机里进行研磨,你就可以得到一瓶辣椒酱了。这意味要制成一瓶辣椒酱,你需要研磨所有的原料。因此,研磨机通常将map操作的蔬菜碎聚集在了一起。

    妻子: 所以,这就是MapReduce?
    我: 你可以说是,也可以说不是。 其实这只是MapReduce的一部分,MapReduce的强大在于分布式计算。
    妻子: 分布式计算? 那是什么?请给我解释下吧。
    我: 假设你参加了一个辣椒酱比赛并且你的食谱赢得了最佳辣椒酱奖。得奖之后,辣椒酱食谱大受欢迎,于是你想要开始出售自制品牌的辣椒酱。假设你每天需要生产10000瓶辣椒酱,你会怎么办呢?
    妻子: 我会找一个能为我大量提供原料的供应商。
    我:是的,就是那样的。那你能否独自完成制作呢?也就是说,独自将原料都切碎? 仅仅一部研磨机又是否能满足需要?而且现在,我们还需要供应不同种类的辣椒酱,像洋葱辣椒酱、青椒辣椒酱、番茄辣椒酱等等。
    妻子: 当然不能了,我会雇佣更多的工人来切蔬菜。我还需要更多的研磨机,这样我就可以更快地生产辣椒酱了。
    我:没错,所以现在你就不得不分配工作了,你将需要几个人一起切蔬菜。每个人都要处理满满一袋的蔬菜,而每一个人都相当于在执行一个简单的Map操作。每一个人都将不断地从袋子里拿出蔬菜来,并且每次只对一种蔬菜进行处理,也就是将它们切碎,直到袋子空了为止。这样,当所有的工人都切完,工作台上就有了洋葱块、番茄块、和蒜蓉等等。
    妻子:但是我怎么会制造出不同种类的番茄酱呢?
    我:现在你会看到MapReduce遗漏的阶段——搅拌。MapReduce将所有输出的蔬菜碎都搅拌在了一起,这些蔬菜碎都是在以key为基础的map操作下产生的。搅拌将自动完成,你可以假设key是一种原料的名字,就像洋葱一样。 所以全部的洋葱keys都会搅拌在一起,并转移到研磨洋葱的研磨器里。这样,你就能得到洋葱辣椒酱了。同样地,所有的番茄也会被转移到标记着番茄的研磨器里,并制造出番茄辣椒酱。

    (4)上面都是从理论上来说明什么是MapReduce,那么咱们在MapReduce产生的过程和代码的角度来理解这个问题。
    如果想统计下过去10年计算机论文出现最多的几个单词,看看大家都在研究些什么,那收集好论文后,该怎么办呢? 

    方法一:可以写一个小程序,把所有论文按顺序遍历一遍,统计每一个遇到的单词的出现次数,最后就可以知道哪几个单词最热门了。 这种方法在数据集比较小时,是非常有效的,而且实现最简单,用来解决这个问题很合适。 

    方法二:写一个多线程程序,并发遍历论文。 这个问题理论上是可以高度并发的,因为统计一个文件时不会影响统计另一个文件。当我们的机器是多核或者多处理器,方法二肯定比方法一高效。但是写一个多线程程序要比方法一困难多了,我们必须自己同步共享数据,比如要防止两个线程重复统计文件。 

    方法三:把作业交给多个计算机去完成。 我们可以使用方法一的程序,部署到N台机器上去,然后把论文集分成N份,一台机器跑一个作业。这个方法跑得足够快,但是部署起来很麻烦,我们要人工把程序copy到别的机器,要人工把论文集分开,最痛苦的是还要把N个运行结果进行整合(当然我们也可以再写一个程序)。 

    方法四:让MapReduce来帮帮我们吧!MapReduce本质上就是方法三,但是如何拆分文件集,如何copy程序,如何整合结果这些都是框架定义好的。我们只要定义好这个任务(用户程序),其它都交给MapReduce。

    map函数和reduce函数:map函数和reduce函数是交给用户实现的,这两个函数定义了任务本身。

    map函数:接受一个键值对(key-value pair),产生一组中间键值对。MapReduce框架会将map函数产生的中间键值对里键相同的值传递给一个reduce函数。 

    reduce函数:接受一个键key,以及相关的一组值(value list),将这组值进行合并产生一组规模更小的值(通常只有一个或零个值)。 

    统计词频的MapReduce函数的核心代码非常简短,主要就是实现这两个函数。

    map(String key, String value):
            // key: document name
            // value: document contents 
            for each word w in value: 
                    EmitIntermediate(w, "1"); 
    reduce(String key, Iterator values): 
         // key: a word 
         // values: a list of counts 
         int result = 0; 
         for each v in values:
             result += ParseInt(v); 
         Emit(AsString(result)); 

      在统计词频的例子里,map函数接受的键(key)是文件名,值(value)是文件的内容,map逐个遍历单词,每遇到一个单词word就产生一个中间键值对<w, "1">(表示单词w咱又找到了一个);MapReduce将键相同(都是单词w)的键值对传给reduce函数,这样reduce函数接受的键就是单词w,值是一串"1"(最基本的实现是这样,但可以优化),个数等于键为w的键值对的个数,然后将这些“1”累加就得到单词w的出现次数。最后这些单词的出现次数会被写到用户定义的位置,存储在底层的分布式存储系统(GFS或HDFS)。 

    【MapReduce工作原理】


    上图是论文里给出的MapReduce流程图。一切都是从最上方的user program开始的,user program链接了MapReduce库,实现了最基本的Map函数和Reduce函数。图中执行的顺序都用数字标记了。

    (1)MapReduce库先把user program的输入文件划分为M份(M为用户定义),每一份通常有16MB到64MB,如图左方所示分成了split0~split4;然后使用fork将用户进程拷贝到集群内其它机器上。

    (2)user program的副本中有一个称为master,其余称为worker,master是负责调度的,为空闲worker分配作业(Map作业或者Reduce作业),worker的数量也是可以由用户指定的。 

    (3)被分配了Map作业的worker,开始读取对应分片的输入数据,Map作业数量是由M决定的,和split一一对应;Map作业从输入数据中抽取出键值对,每一个键值对都作为参数传递给map函数,map函数产生的中间键值对被缓存在内存中(环形缓冲区kvBuffer)。

    (4)缓存的中间键值对会被定期写入本地磁盘(spill),而且被分为R个区,R的大小是由用户定义的,将来每个区会对应一个Reduce作业;这些中间键值对的位置会被通报给master,master负责将信息转发给Reduce worker。 

    (5)master通知分配了Reduce作业的worker它负责的分区在什么位置(肯定不止一个地方,每个Map作业产生的中间键值对都可能映射到所有R个不同分区),当Reduce worker把所有它负责的中间键值对都读过来后,先对它们进行排序,使得相同键的键值对聚集在一起。因为不同的键可能会映射到同一个分区也就是同一个Reduce作业,所以排序是必须的。 

    (6)reduce worker遍历排序后的中间键值对,对于每个唯一的键,都将键与关联的值传递给reduce函数,reduce函数产生的输出会添加到这个分区的输出文件中。

    (7)当所有的Map和Reduce作业都完成了,master唤醒正版的user program,MapReduce函数调用返回user program的代码。 

    所有执行完毕后,MapReduce输出放在了R个分区的输出文件中(分别对应一个Reduce作业)。用户通常并不需要合并这R个文件,而是将其作为输入交给另一个MapReduce程序处理。整个过程中,输入数据来自底层分布式文件系统(hdfs),中间数据是放在本地文件系统的,最终输出数据是写入hdfs的。注意Map/Reduce作业和map/reduce函数的区别:Map作业处理一个输入数据的分片,可能需要调用多次map函数来处理每个输入键值对;Reduce作业处理一个分区的中间键值对,期间要对每个不同的键调用一次reduce函数,Reduce作业最终也对应一个输出文件。

    【总结】

    通过以上你是否了解什么是MapReduce了,什么是key,怎么过滤有效数据,怎么得到自己想要的数据。

    MapReduce是一种编程思想,可以使用java来实现,C++来实现。Map的作用是过滤一些原始数据,Reduce则是处理这些数据,得到我们想要的结果(比如造出番茄辣椒酱)。也就是我们使用hadoop,比如进行日志处理之后,得到我们关心的数据。



    展开全文
  • 学习MapReduce?这一篇就够了

    千次阅读 多人点赞 2018-09-01 08:18:08
    1 MapReduce原理 1.1 为什么要MapReduce 1.2 MapReduce框架结构及核心运行机制 1.2.1 结构 1.2.2 MR程序运行流程 1.3 MapTask并行度决定机制 1.3.1 mapTask并行度的决定机制 1.3.2 FileInputFormat切片机制 ...
    1. 本博客已迁移至微信公众号!将不再更新
    2. 关注公众号即可获得免费学习资源,获得免费指导!!!
    3. 公众号后续将会持续更新clickhouse,sparkstreaming,flink,数仓建模,用户画像,实时计算,推荐系统,实时数仓等内容,感兴趣的朋友可以关注
    4. 不定期会有朋友的面经分享

     

    目录

    1 MapReduce原理

    1.1 为什么要MapReduce

    1.2 MapReduce框架结构及核心运行机制

    1.2.1 结构

    1.2.2 MR程序运行流程

    1.3 MapTask并行度决定机制

    1.3.1 mapTask并行度的决定机制

    1.3.2 FileInputFormat切片机制

    1.4 ReduceTask并行度的决定

    1.5MapReduce程序演示

    2 MapReduce实践篇

    2.1 mapreduce示例编写及编程规范

    2.1.1 编程规范

    2.1.2 wordcount示例编写

    2.2 mapreduce程序运行模式

    2.2.1 本地运行模式

    2.2.2 集群运行模式

    2.3 MapReduce中的Combiner

    3 MapReduce原理篇

    3.1 mapreduce的shuffle机制

    3.1.1 概述:

    3.1.2 主要流程

    3.1.3 详细流程

    3.2 mapreduce中的序列化

    3.2.1 概述

    3.2.2 自定义对象实现MR中的序列化接口

    3.3 mapreduce与yarn

    3.3.1 yarn概述

    3.3.2 yarn的重要概念

    4 mapreduce案例(mapreduce中的排序)

    4.1 需求

    4.2 分析

    4.3 实现


    1 MapReduce原理

    Mapreduce是一个分布式运算程序的编程框架是用户开发“基于hadoop的数据分析应用”的核心框架;

    Mapreduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序,并发运行在一个hadoop集群上;

    1.1 为什么要MapReduce

    (1)海量数据在单机上处理因为硬件资源限制,无法胜任

    (2)而一旦将单机版程序扩展到集群来分布式运行,将极大增加程序的复杂度和开发难度

    (3)引入mapreduce框架后,开发人员可以将绝大部分工作集中在业务逻辑的开发上,而将分布式计算中的复杂性交由框架来处理

    1.2 MapReduce框架结构及核心运行机制

    1.2.1 结构

    一个完整的mapreduce程序在分布式运行时有三类实例进程:

    1、MRAppMaster:负责整个程序的过程调度及状态协调

    2、mapTask:负责map阶段的整个数据处理流程

    3、ReduceTask:负责reduce阶段的整个数据处理流程

    1.2.2 MR程序运行流程

    1.2.2.1 流程示意图

    1.2.2.2 流程解析

    1.一个mr程序启动的时候,最先启动的是MRAppMaster,MRAppMaster启动后根据本次job的描述信息,计算出需要的maptask实例数量,然后向集群申请机器启动相应数量的maptask进程

    2.maptask进程启动之后,根据给定的数据切片范围进行数据处理,主体流程为:

    • 利用客户指定的inputformat来获取RecordReader读取数据,形成输入KV对
    • 将输入KV对传递给客户定义的map()方法,做逻辑运算,并将map()方法输出的KV对收集到缓存
    • 将缓存中的KV对按照K分区排序后不断溢写到磁盘文件

    3.MRAppMaster监控到所有maptask进程任务完成之后,会根据客户指定的参数启动相应数量的reducetask进程,并告知reducetask进程要处理的数据范围(数据分区)

    4.Reducetask进程启动之后,根据MRAppMaster告知的待处理数据所在位置,从若干台maptask运行所在机器上获取到若干个maptask输出结果文件,并在本地进行重新归并排序,然后按照相同key的KV为一个组,调用客户定义的reduce()方法进行逻辑运算,并收集运算输出的结果KV,然后调用客户指定的outputformat将结果数据输出到外部存储

    1.3 MapTask并行度决定机制

    maptask的并行度决定map阶段的任务处理并发度,进而影响到整个job的处理速度

    那么,mapTask并行实例是否越多越好呢?其并行度又是如何决定呢?

    1.3.1 mapTask并行度的决定机制

    一个job的map阶段并行度由客户端在提交job时决定

    而客户端对map阶段并行度的规划的基本逻辑为:

    将待处理数据执行逻辑切片(即按照一个特定切片大小,将待处理数据划分成逻辑上的多个split),然后每一个split分配一个mapTask并行实例处理

    这段逻辑及形成的切片规划描述文件,由FileInputFormat实现类的getSplits()方法完成,其过程如下图:

    1.3.2 FileInputFormat切片机制

    1、切片定义在InputFormat类中的getSplit()方法

    2、FileInputFormat中默认的切片机制:

    • 简单地按照文件的内容长度进行切片
    • 切片大小,默认等于block大小
    • 切片时不考虑数据集整体,而是逐个针对每一个文件单独切片

    比如待处理数据有两个文件:

    file1.txt    320M

    file2.txt    10M

    经过FileInputFormat的切片机制运算后,形成的切片信息如下:  

    file1.txt.split1--  0~128

    file1.txt.split2--  128~256

    file1.txt.split3--  256~320

    file2.txt.split1--  0~10M

    3、FileInputFormat中切片的大小的参数配置

    通过分析源码,在FileInputFormat中,计算切片大小逻辑Math.max(minSize, Math.min(maxSize, blockSize));  切片主要由这几个值来运算决定

    minsize:默认值:1  

       配置参数: mapreduce.input.fileinputformat.split.minsize    

    maxsize:默认值:Long.MAXValue  

        配置参数:mapreduce.input.fileinputformat.split.maxsize

    blocksize

    因此,默认情况下,切片大小=blocksize

    maxsize(切片最大值):

    参数如果调得比blocksize小,则会让切片变小,而且就等于配置的这个参数的值

    minsize (切片最小值):

    参数调的比blockSize大,则可以让切片变得比blocksize还大

    选择并发数的影响因素:

    1. 运算节点的硬件配置
    2. 运算任务的类型:CPU密集型还是IO密集型
    3. 运算任务的数据量

    1.4 ReduceTask并行度的决定

    reducetask的并行度同样影响整个job的执行并发度和执行效率,但与maptask的并发数由切片数决定不同,Reducetask数量的决定是可以直接手动设置:

    //默认值是1,手动设置为4

    job.setNumReduceTasks(4);

    如果数据分布不均匀,就有可能在reduce阶段产生数据倾斜

    注意: reducetask数量并不是任意设置,还要考虑业务逻辑需求,有些情况下,需要计算全局汇总结果,就只能有1个reducetask

    尽量不要运行太多的reduce task。对大多数job来说,最好rduce的个数最多和集群中的reduce持平,或者比集群的 reduce slots小。这个对于小集群而言,尤其重要。

    1.5MapReduce程序演示

    Hadoop的发布包中内置了一个hadoop-mapreduce-example-2.4.1.jar,这个jar包中有各种MR示例程序,可以通过以下步骤运行:

    启动hdfs,yarn

    然后在集群中的任意一台服务器上启动执行程序(比如运行wordcount):

    hadoop jar hadoop-mapreduce-example-2.4.1.jar wordcount  /wordcount/data /wordcount/out

    2 MapReduce实践篇

    2.1 mapreduce示例编写及编程规范

    2.1.1 编程规范

    1. 用户编写的程序分成三个部分:Mapper,Reducer,Driver(提交运行mr程序的客户端)
    2. Mapper的输入数据是KV对的形式(KV的类型可自定义)
    3. Mapper的输出数据是KV对的形式(KV的类型可自定义)
    4. Mapper中的业务逻辑写在map()方法中
    5. map()方法(maptask进程)对每一个<K,V>调用一次
    6. Reducer的输入数据类型对应Mapper的输出数据类型,也是KV
    7. Reducer的业务逻辑写在reduce()方法中
    8. Reducetask进程对每一组相同k<k,v>组调用一次reduce()方法
    9. 用户自定义的Mapper和Reducer都要继承各自的父类
    10. 整个程序需要一个Drvier来进行提交,提交的是一个描述了各种必要信息的job对象

    2.1.2 wordcount示例编写

    需求:在一堆给定的文本文件中统计输出每一个单词出现的总次数

    (1)定义一个mapper类

    //首先要定义四个泛型的类型
    //keyin:  LongWritable    valuein: Text
    //keyout: Text            valueout:IntWritable
    
    public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
    	//map方法的生命周期:  框架每传一行数据就被调用一次
    	//key :  这一行的起始点在文件中的偏移量
    	//value: 这一行的内容
    	@Override
    	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    		//拿到一行数据转换为string
    		String line = value.toString();
    		//将这一行切分出各个单词
    		String[] words = line.split(" ");
    		//遍历数组,输出<单词,1>
    		for(String word:words){
    			context.write(new Text(word), new IntWritable(1));
    		}
    	}
    }

    将单词作为key,次数1为value。以便于后续的数据分发,根据单词分发/相同的单词分到相同的reduce task

    (2)定义一个reducer类

    //生命周期:框架每传递进来一个kv 组,reduce方法被调用一次
    	@Override
    	protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    		//定义一个计数器
    		int count = 0;
    		//遍历这一组kv的所有v,累加到count中
    		for(IntWritable value:values){
    			count += value.get();
    		}
    		context.write(key, new IntWritable(count));
    	}
    }

    (3)定义一个主类,用来描述job并提交job

    package cn.test.bigdata.mr.wcdemo;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    public class WordcountDriver {
    
    	
    	public static void main(String[] args) throws Exception {
    
    		
    		Configuration conf = new Configuration();
    		
    		//设置的没有用!  ??????
    //		conf.set("HADOOP_USER_NAME", "hadoop");
    //		conf.set("dfs.permissions.enabled", "false");
    		
    		//不提交在yarn上面,只在本地跑
    		conf.set("mapreduce.framework.name", "local");
    		//本地模式运行mr程序时,输入输出的数据可以在本地,也可以在hdfs上
    		//到底在哪里,酒宴以下两行配置,用的是哪行,默认是本地的
    		conf.set("fs.defaultFS", "file:///");
    		/*conf.set("fs.defaultFS", "hdfs://192.168.175.128:9000/");
    		conf.set("mapreduce.framework.name", "yarn");
    		conf.set("yarn.resoucemanager.hostname", "192.168.178.128");*/
    		Job job = Job.getInstance(conf);
    		
    		/*job.setJar("/home/hadoop/wc.jar");*/
    		//指定本程序的jar包所在的本地路径
    		job.setJarByClass(WordcountDriver.class);
    		
    		//指定本业务job要使用的mapper/Reducer业务类
    		job.setMapperClass(WordcountMapper.class);
    		job.setReducerClass(WordcountReduce.class);
    		
    		//指定需要使用combiner,以及用哪个类作为combiner的逻辑
    		/*job.setCombinerClass(WordcountCombiner.class);*/
    		job.setCombinerClass(WordcountReduce.class);
    		
    		//如果不设置InputFormat,它默认用的是TextInputformat.class
    		/*job.setInputFormatClass(CombineTextInputFormat.class);
    		CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);
    		CombineTextInputFormat.setMinInputSplitSize(job, 2097152);*/
    		
    		//指定mapper输出数据的kv类型
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(IntWritable.class);
    		
    		//指定最终输出的数据的kv类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(IntWritable.class);
    		
    		//指定job的输入原始文件所在目录
    		FileInputFormat.setInputPaths(job, new Path("D:\\wordstest\\input"));
    		//指定job的输出结果所在目录
    		FileOutputFormat.setOutputPath(job, new Path("D:\\wordstest\\output"));
    		
    		//将job中配置的相关参数,以及job所用的java类所在的jar包,提交给yarn去运行
    		/*job.submit();*/
    		boolean res = job.waitForCompletion(true);
    		System.exit(res?0:1);
    		
    	}
    	
    
    
    }
    

    2.2 mapreduce程序运行模式

    2.2.1 本地运行模式

    1. mapreduce程序是被提交给LocalJobRunner在本地以单进程的形式运行
    2. 而处理的数据及输出结果可以在本地文件系统,也可以在hdfs上
    3. 怎样实现本地运行?写一个程序,不要带集群的配置文件(本质是你的mr程序的conf中是否有mapreduce.framework.name=local以及yarn.resourcemanager.hostname参数)
    4. 本地模式非常便于进行业务逻辑的debug,只要在eclipse中打断点即可

    如果在windows下想运行本地模式来测试程序逻辑,需要在windows中配置环境变量:

    %HADOOP_HOME%  =  d:/hadoop-2.6.1

    %PATH% =  %HADOOP_HOME%\bin

    并且要将d:/hadoop-2.6.1的lib和bin目录替换成windows平台编译的版本

    2.2.2 集群运行模式

    1. mapreduce程序提交给yarn集群resourcemanager,分发到很多的节点上并发执行
    2. 处理的数据和输出结果应该位于hdfs文件系统(也需要启动yarn)
    3. 提交集群的实现步骤:需要在输入路径有文件

    A、将程序打成JAR包,然后在集群的任意一个节点上用hadoop命令启动

         $ hadoop jar wordcount.jar cn.test.bigdata.mrsimple.WordCountDriver inputpath outputpath

    B、直接在linux的eclipse中运行main方法

    (项目中要带参数:mapreduce.framework.name=yarn以及yarn的两个基本配置)

    C、如果要在windows的eclipse中提交job给集群,则要修改YarnRunner类

     

    2.3 MapReduce中的Combiner

    1. combiner是MR程序中Mapper和Reducer之外的一种组件
    2. combiner组件的父类就是Reducer
    3. combiner和reducer的区别在于运行的位置:

             Combiner是在每一个maptask所在的节点运行

             Reducer是接收全局所有Mapper的输出结果;

    (4) combiner的意义就是对每一个maptask的输出进行局部汇总,以减小网络传输量

    具体实现步骤:

    1. 自定义一个combiner继承Reducer,重写reduce方法
    2. 在job中设置:  job.setCombinerClass(CustomCombiner.class)

    (5) combiner能够应用的前提是不能影响最终的业务逻辑

    而且,combiner的输出kv应该跟reducer的输入kv类型要对应起来

     

    3 MapReduce原理篇

    3.1 mapreduce的shuffle机制

    3.1.1 概述:

    1. mapreduce中,map阶段处理的数据如何传递给reduce阶段,是mapreduce框架中最关键的一个流程,这个流程就叫shuffle;
    2. shuffle: 洗牌、发牌——(核心机制:数据分区,排序,缓存)
    3. 具体来说:就是将maptask输出的处理结果数据,分发给reducetask,并在分发的过程中,对数据按key进行了分区和排序

    3.1.2 主要流程

    shuffle是MR处理流程中的一个过程,它的每一个处理步骤是分散在各个map task和reduce task节点上完成的,整体来看,分为3个操作:

    1. 分区partition
    2. Sort根据key排序
    3. Combiner进行局部value的合并

    3.1.3 详细流程

    1. maptask收集我们的map()方法输出的kv对,放到内存缓冲区中
    2. 从内存缓冲区不断溢出本地磁盘文件,可能会溢出多个文件 (默认100M)
    3. 多个溢出文件会被合并成大的溢出文件
    4. 在溢出过程中,及合并的过程中,都要调用partitoner进行分组和针对key进行排序
    5. reducetask根据自己的分区号,去各个maptask机器上相应的结果分区数据
    6. reducetask会取到同一个分区的来自不同maptask的结果文件,reducetask会将这些文件再进行合并(归并排序)
    7. 合并成大文件后,shuffle的过程也就结束了,后面进入reducetask的逻辑运算过程(从文件中取出一个一个的键值对group,调用用户自定义的reduce()方法)

    Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率,原则上说,缓冲区越大,磁盘io的次数越少,执行速度就越快

    缓冲区的大小可以通过参数调整,  参数:io.sort.mb  默认100M

     

    3.2 mapreduce中的序列化

    3.2.1 概述

    Java的序列化是一个重量级序列化框架(Serializable),一个对象被序列化后,会附带很多额外的信息(各种校验信息,header,继承体系。。。。),不便于在网络中高效传输;

    所以,hadoop自己开发了一套序列化机制(Writable),精简,高效

    3.2.2 自定义对象实现MR中的序列化接口

    如果需要将自定义的bean放在key中传输,则还需要实现comparable接口,因为mapreduce框中的shuffle过程一定会对key进行排序,此时,自定义的bean实现的接口应该是:

    public  class  FlowBean  implements  WritableComparable<FlowBean>

    需要自己实现的方法是:

    /**
    	 * 反序列化的方法,反序列化时,从流中读取到的各个字段的顺序应该与序列化时写出去的顺序保持一致
    	 */
    	@Override
    	public void readFields(DataInput in) throws IOException {
    		
    		upflow = in.readLong();
    		dflow = in.readLong();
    		sumflow = in.readLong();
    		
    
    	}
    
    	/**
    	 * 序列化的方法
    	 */
    	@Override
    	public void write(DataOutput out) throws IOException {
    
    		out.writeLong(upflow);
    		out.writeLong(dflow);
    		//可以考虑不序列化总流量,因为总流量是可以通过上行流量和下行流量计算出来的
    		out.writeLong(sumflow);
    
    	}
    	
    	@Override
    	public int compareTo(FlowBean o) {
    		
    		//实现按照sumflow的大小倒序排序
    		return sumflow>o.getSumflow()?-1:1;
    	}

    3.3 mapreduce与yarn

    3.3.1 yarn概述

    Yarn是一个资源调度平台,负责为运算程序提供服务器运算资源,相当于一个分布式的操作系统平台,而mapreduce等运算程序则相当于运行于操作系统之上的应用程序

    3.3.2 yarn的重要概念

    1. yarn并不清楚用户提交的程序的运行机制
    2. yarn只提供运算资源的调度(用户程序向yarn申请资源,yarn就负责分配资源)
    3. yarn中的主管角色叫ResourceManager
    4. yarn中具体提供运算资源的角色叫NodeManager
    5. 这样一来,yarn其实就与运行的用户程序完全解耦,就意味着yarn上可以运行各种类型的分布式运算程序(mapreduce只是其中的一种),比如mapreduce、storm程序,spark程序,tez ……
    6. 所以,spark、storm等运算框架都可以整合在yarn上运行,只要他们各自的框架中有符合yarn规范的资源请求机制即可
    7. Yarn就成为一个通用的资源调度平台,从此,企业中以前存在的各种运算集群都可以整合在一个物理集群上,提高资源利用率,方便数据共享

    4 mapreduce案例(mapreduce中的排序)

    4.1 需求

    对日志数据中的上下行流量信息汇总,并输出按照总流量倒序排序的结果

    数据如下:

    1363157985066 13726230503 00-FD-07-A4-72-B8:CMCC 120.196.100.82 24 27 2481 24681 200

    1363157995052 13826544101 5C-0E-8B-C7-F1-E0:CMCC 120.197.40.4 4 0 264 0 200

    1363157991076 13926435656 20-10-7A-28-CC-0A:CMCC 120.196.100.99 2 4 132 1512 200

    1363154400022 13926251106 5C-0E-8B-8B-B1-50:CMCC 120.197.40.4 4 0 240 0 200

    4.2 分析

    基本思路:实现自定义的bean来封装流量信息,并将bean作为map输出的key来传输

    MR程序在处理数据的过程中会对数据排序(map输出的kv对传输到reduce之前,会排序),排序的依据是map输出的key

    所以,我们如果要实现自己需要的排序规则,则可以考虑将排序因素放到key中,让key实现接口:WritableComparable

    然后重写key的compareTo方法

    4.3 实现

    1. 自定义bean

    public class FlowBean implements WritableComparable<FlowBean>{
    	
    	long upflow;
    	long downflow;
    	long sumflow;
    	
    	//如果空参构造函数被覆盖,一定要显示定义一下,否则在反序列时会抛异常
    	public FlowBean(){}
    	
    	public FlowBean(long upflow, long downflow) {
    		super();
    		this.upflow = upflow;
    		this.downflow = downflow;
    		this.sumflow = upflow + downflow;
    	}
    	
    	public long getSumflow() {
    		return sumflow;
    	}
    
    	public void setSumflow(long sumflow) {
    		this.sumflow = sumflow;
    	}
    
    	public long getUpflow() {
    		return upflow;
    	}
    	public void setUpflow(long upflow) {
    		this.upflow = upflow;
    	}
    	public long getDownflow() {
    		return downflow;
    	}
    	public void setDownflow(long downflow) {
    		this.downflow = downflow;
    	}
    
    	//序列化,将对象的字段信息写入输出流
    	@Override
    	public void write(DataOutput out) throws IOException {
    		
    		out.writeLong(upflow);
    		out.writeLong(downflow);
    		out.writeLong(sumflow);
    		
    	}
    
    	//反序列化,从输入流中读取各个字段信息
    	@Override
    	public void readFields(DataInput in) throws IOException {
    		upflow = in.readLong();
    		downflow = in.readLong();
    		sumflow = in.readLong();
    		
    	}

    2 mapper和reducer

    public class FlowCount {
    
    	static class FlowCountMapper extends Mapper<LongWritable, Text, FlowBean,Text > {
    
    		@Override
    		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    
    			String line = value.toString();
    			String[] fields = line.split("\t");
    			try {
    				String phonenbr = fields[0];
    
    				long upflow = Long.parseLong(fields[1]);
    				long dflow = Long.parseLong(fields[2]);
    
    				FlowBean flowBean = new FlowBean(upflow, dflow);
    
    				context.write(flowBean,new Text(phonenbr));
    			} catch (Exception e) {
    
    				e.printStackTrace();
    			}
    
    		}
    
    	}
    
    	static class FlowCountReducer extends Reducer<FlowBean,Text,Text, FlowBean> {
    
    		@Override
    		protected void reduce(FlowBean bean, Iterable<Text> phonenbr, Context context) throws IOException, InterruptedException {
    
    			Text phoneNbr = phonenbr.iterator().next();
    
    			context.write(phoneNbr, bean);
    
    		}
    
    	}
    
    	public static void main(String[] args) throws Exception {
    
    		Configuration conf = new Configuration();
    
    		Job job = Job.getInstance(conf);
    
    		job.setJarByClass(FlowCount.class);
    
    		job.setMapperClass(FlowCountMapper.class);
    		job.setReducerClass(FlowCountReducer.class);
    
    		 job.setMapOutputKeyClass(FlowBean.class);
    		 job.setMapOutputValueClass(Text.class);
    
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(FlowBean.class);
    
    		// job.setInputFormatClass(TextInputFormat.class);
    
    		FileInputFormat.setInputPaths(job, new Path(args[0]));
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
    		job.waitForCompletion(true);
    
    	}
    
    }

     

     

     

     

     

    展开全文
  • 一、MapReduce的概念 MapReduce是一种编程模型,用于大规模数据集(大于1TB)的并行运算。概念"Map(映射)"和"Reduce(归约)",和它们的主要思想,都是从函数式编程语言里借来的,还有从矢量...

    一、MapReduce的概念

    MapReduce是一种编程模型,用于大规模数据集(大于1TB)的并行运算。概念"Map(映射)"和"Reduce(归约)",和它们的主要思想,都是从函数式编程语言里借来的,还有从矢量编程语言里借来的特性。它极大地方便了编程人员在不会分布式并行编程的情况下,将自己的程序运行在分布式系统上。 当前的软件实现是指定一个Map(映射)函数,用来把一组键值对映射成一组新的键值对,指定并发的Reduce(归约)函数,用来保证所有映射的键值对中的每一个共享相同的键组。

    记住这2点:

          1、MapReduce是一种分布式计算模型,是Google提出的,主要用于搜索领域,解决海量数据的计算问题。

          2、MR有两个阶段组成:Map和Reduce,用户只需实现map()和reduce()两个函数,即可实现分布式计算。

    二、MapReduce的工作流程

                                                    

    下面对上面出现的一些名词进行介绍:

    ResourceManager:是YARN资源控制框架的中心模块,负责集群中所有的资源的统一管理和分配。它接收来自NM(NodeManager)的汇报,建立AM,并将资源派送给AM(ApplicationMaster)。

    NodeManager:简称NM,NodeManager是ResourceManager在每台机器的上代理,负责容器的管理,并监控他们的资源使用情况(cpu,内存,磁盘及网络等),以及向 ResourceManager提供这些资源使用报告。

    ApplicationMaster:以下简称AM。YARN中每个应用都会启动一个AM,负责向RM申请资源,请求NM启动container,并告诉container做什么事情。

    Container:资源容器。YARN中所有的应用都是在container之上运行的。AM也是在container上运行的,不过AM的container是RM申请的。

        (1)  Container是YARN中资源的抽象,它封装了某个节点上一定量的资源(CPU和内存两类资源)。

        (2) Container由ApplicationMaster向ResourceManager申请的,由ResouceManager中的资源调度器异步分配给ApplicationMaster;

         (3)Container的运行是由ApplicationMaster向资源所在的NodeManager发起的,Container运行时需提供内部执行的任务命令(可以是任何命令,比如java、Python、C++进程启动命令均可)以及该命令执行所需的环境变量和外部资源(比如词典文件、可执行文件、jar包等)。

    另外,一个应用程序所需的Container分为两大类,如下:

        (1) 运行ApplicationMaster的Container:这是由ResourceManager(向内部的资源调度器)申请和启动的,用户提交应用程序时,可指定唯一的ApplicationMaster所需的资源;

         (2)运行各类任务的Container:这是由ApplicationMaster向ResourceManager申请的,并由ApplicationMaster与NodeManager通信以启动之。

    以上两类Container可能在任意节点上,它们的位置通常而言是随机的,即ApplicationMaster可能与它管理的任务运行在一个节点上。

    整个MapReduce的过程大致分为 Map-->Shuffle(排序)-->Combine(组合)-->Reduce

    三、MapReduce流程分析

    1、Map端:

    (1).每个输入分片会让一个map任务来处理,默认情况下,以HDFS的一个块的大小(默认为64M)为一个分片,当然我们也可以设置块的大小。map输出的结果会暂且放在一个环形内存缓冲区中(该缓冲区的大小默认为100M,由io.sort.mb属性控制),当该缓冲区快要溢出时(默认为缓冲区大小的80%,由io.sort.spill.percent属性控制),会在本地文件系统中创建一个溢出文件,将该缓冲区中的数据写入这个文件。

    (2).在写入磁盘之前,线程首先根据reduce任务的数目将数据划分为相同数目的分区,也就是一个reduce任务对应一个分区的数据。这样做是为了避免有些reduce任务分配到大量数据,而有些reduce任务却分到很少数据,甚至没有分到数据的尴尬局面。其实分区就是对数据进行hash的过程。然后对每个分区中的数据进行排序,如果此时设置了Combiner,将排序后的结果进行Combia操作,这样做的目的是让尽可能少的数据写入到磁盘。

    (3).当map任务输出最后一个记录时,可能会有很多的溢出文件,这时需要将这些文件合并。合并的过程中会不断地进行排序和combia操作,目的有两个:1.尽量减少每次写入磁盘的数据量;2.尽量减少下一复制阶段网络传输的数据量。最后合并成了一个已分区且已排序的文件。为了减少网络传输的数据量,这里可以将数据压缩,只要将mapred.compress.map.out设置为true就可以了。

    (4).将分区中的数据拷贝给相对应的reduce任务。有人可能会问:分区中的数据怎么知道它对应的reduce是哪个呢?其实map任务一直和其父TaskTracker保持联系,而TaskTracker又一直和JobTracker保持心跳。所以JobTracker中保存了整个集群中的宏观信息。只要reduce任务向JobTracker获取对应的map输出位置就ok了哦。

    到这里,map端就分析完了。那到底什么是Shuffle呢?Shuffle的中文意思是“洗牌”,如果我们这样看:一个map产生的数据,结果通过hash过程分区却分配给了不同的reduce任务,是不是一个对数据洗牌的过程呢?呵呵。

    2、Reduce端:

    (1).Reduce会接收到不同map任务传来的数据,并且每个map传来的数据都是有序的。如果reduce端接受的数据量相当小,则直接存储在内存中(缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制,表示用作此用途的堆空间的百分比),如果数据量超过了该缓冲区大小的一定比例(由mapred.job.shuffle.merge.percent决定),则对数据合并后溢写到磁盘中。

    (2).随着溢写文件的增多,后台线程会将它们合并成一个更大的有序的文件,这样做是为了给后面的合并节省时间。其实不管在map端还是reduce端,MapReduce都是反复地执行排序,合并操作,现在终于明白了有些人为什么会说:排序是hadoop的灵魂。

    (3).合并的过程中会产生许多的中间文件(写入磁盘了),但MapReduce会让写入磁盘的数据尽可能地少,并且最后一次合并的结果并没有写入磁盘,而是直接输入到reduce函数。

    人工智能、大数据、云计算和物联网的未来发展值得重视,均为前沿产业,有兴趣的朋友,可以查阅多智时代,在此为你推荐几篇优质好文:

    1. Hadoop新一代MapReduce框架更快、更强
    2. Hadoop有哪些优势,发展趋势和前景是什么?
    3. Hadoop和Spark之间有什么区别,现工业界大数据技术都在使用何种技术

     

     

    展开全文
  • MapReduce介绍

    万次阅读 多人点赞 2018-10-21 12:45:11
    MapReduce产生背景 如果让你统计日志里面的出现的某个URL的总次数,让你自己去写个单机版的程序,写个逻辑:无非就是读这个文件一行,然后把那个地方截取出来,截取出来之后,然后可以把它放到一个HashMap里面...
  • MapReduce全过程

    千次阅读 2019-02-25 17:42:11
    而在MapReduce中,shuffle更像是洗牌的逆过程,指的是将map端的无规则输出按指定的规则“打乱”成具有一定规则的数据,以便reduce端接收处理。其在MapReduce中所处的工作阶段是map输出后到reduce接收前,具体可以...
  • MapReduce中的map与reduce

    千次阅读 2019-02-12 09:19:47
    MapReduce中的map与reduce     本文主要介绍MapReduce的map与reduce所包含的各各阶段  MapReduce中的每个map任务可以细分4个阶段:record reader、mapper、combiner和partitioner。map...
  • 谷歌三大论文之一-MapReduce

    千次阅读 2017-08-20 18:55:01
    MapReduce是一个编程模型,也是一个处理和生成超大数据集的算法模型的相关实现。用户首先创建一个Map函数处理一个基于 key/value pair的数据集合,输出中间的基于key/value pair的数据集合;然后再创建一个Reduce...
  • 用通俗易懂的大白话讲解Map/Reduce原理

    万次阅读 多人点赞 2014-03-31 16:20:44
    下面是我自己的微信公众号(不定期更新 JAVA 、大数据、个人成长等干货) ...Hadoop简介Hadoop就是一个实现了Google云计算系统的开源系统,包括并行计算模型Map/Reduce,分布式文件系统HDFS,以及...
  • Shuffle过程是MapReduce的核心,也被称为奇迹发生的地方。要想理解MapReduce, Shuffle是必须要了解的。Shuffle的正常意思是洗牌或弄乱,可能大家更熟悉的是Java API里Collections.shuffle(List)方法,它会随机地...
  • 什么是Map Reduce

    2018-07-18 15:11:15
    什么是Map/Reduce,看下面的各种解释: (1)MapReduce是hadoop的核心组件之一,hadoop要分布式包括两部分,一是分布式文件系统hdfs,一部是分布式计算框,就是mapreduce,缺一不可,也就是说,可以通过mapreduce很容易...
  • MapReduce之:大白话讲解Map/Reduce原理

    千次阅读 2018-05-10 11:31:41
    Hadoop简介Hadoop就是一个实现了Google云计算系统的开源系统,包括并行计算模型Map/Reduce,分布式文件系统HDFS,以及分布式数据库Hbase,同时Hadoop的相关项目也很丰富,包括ZooKeeper,Pig,Chukwa,Hive,Hbase,...
  • MapreduceMapReduce任务的个数

    万次阅读 2018-04-11 16:30:11
    1、Map任务的个数 读取数据产生多少个Mapper?? Mapper数据过大的话,会产生大量的小文件,过多的Mapper创建和初始化都会消耗大量的硬件资源 Mapper数太小,并发度过小,Job执行时间过长,无法充分利用分布式硬件...
  • Java操作MongoDB Map Reduce的使用

    千次阅读 2017-07-03 10:28:06
    因工作需要,把数据库中的旧数据导出并以JSON格式存到本地。考虑到数据量已经达到了十几个G,因此用sql直接查询显然是行不通的。 所以就想到了使用MongoDB,关于MongoDB的安装,Mongo Shell的命令以及基本增删改查...
  • MapReduce中的MapReduce个数设定

    千次阅读 2018-07-03 15:14:55
    一、Map的个数  在map阶段读取数据前,FileInputFormat会将输入文件分割成spilt,而spilt的个数决定了map的个数(一个spilt分片对应一个map)。影响map个数的因素只要有:  1)文件的大小。比如,当文件大于128M...
  • 实现代码如下: public class MyList { public static void main(String[] args) { List&lt;Integer&gt; aa =new ArrayList&lt;&gt;(); aa.add(1); aa.add(2); aa.add(3);... ...
  • MapReduce详解

    万次阅读 多人点赞 2017-12-11 09:26:41
    1.1 MapReduce是什么  Hadoop MapReduce是一个软件框架,基于该框架能够容易地编写应用程序,这些应用程序能够运行在由上千个商用机器组成的大集群上,并以一种可靠的,具有容错能力的方式并行地处理上TB级别的...
  • hive map reduce 参数设置

    千次阅读 2018-03-12 16:51:06
    现象:1....3.不少sql执行时的map数量能达到1000-3000,但是reduce数量在1左右。由于每天load出的excle文档,sjzx租户的sql过长,手动无法执行,其它租户的sql抽取时又存在乱码问题,仅针对了sjzx_b...
  • MapReduce中Shuffle机制详解——Map端Shuffle链接 2、Reduce端shuffle Reduce端的shuffle主要包括三个阶段,copy,sort(merge),reduce Map的输出文件放置在运行MapTask的NodeManager的本地磁盘上,它是运行...
  • MongoDB Map Reduce

    2016-09-09 21:24:34
    Map-Reduce是一种计算模型,简单的说是将大量的工作(数据)分解(MAP)执行,然后再将结果合并成最终结果(RECUCE). MongoDB提供的Map-Redeuce非常灵活,对于大规模数据分析也相当实用.MapReduce命令以下是MapRedcuce的...
  • mapreduce 作业中 map/reduce 个数的计算

    千次阅读 2018-10-08 16:59:40
     当在hadoop集群提交mapreduce作业时,mapreduce 的个数是如何计算的?   2. map个数的计算  2.1 map个数的计算和分片大小(splitSize)是有关系的,所以我们先看看splitSize的计算公式: long ...
  • mapreduce是hadoop的核心功能,hadoop正是通过多个mapreduce的并行运行来实现任务的分布式并行计算,从这个观点来看,如果将mapreduce的数量设置为1,那么用户的任务就没有并行执行,但是mapreduce的数量也...
  • Map 的数量通常是由 Hadoop 集群的数据块大小(输入文件的...(1)如果任务数是节点个数的0.95倍,那么所有的Reduce任务能够在 Map任务的输出传输结束后同时开始运行。(2)如果任务数是节点个数的1.75倍,那么高速
  •  1 使用Java编写MapReduce程序时,如何向mapreduce函数传递参数。  2 使用Streaming编写MapReduce程序(C/C++, Shell, Python)时,如何向mapreduce脚本传递参数。  3 使用Streaming编写MapReduce程序(C/C++, ...
  • 深入理解与应用Hadoop中的MapReduce

    万次阅读 2018-11-28 19:23:26
    现在大数据是越来越火了,而我自己研究这...1、本篇博客核心内容:2、MR的基本概念3、MR中map()函数和reduce()函数如何编写4、MR程序的基本编写流程(MR的基本执行过程) 下面将用一个具体的电信业务说明MR最基本的编写
  • MapTask运行机制详解 整个Map阶段流程大体如图所示 简单概述 inputFile通过split被逻辑切分为多个split文件, 通过Record按行读取内容给map(用户自己实现的)进行处理, 数据被map处理结束之后交给OutputCollector...
  • Hive是基于Hadoop的一个数据仓库系统,在各大公司都有广泛的应用。美团数据仓库也是基于Hive搭建,每天执行近万次的Hive ETL...通过向社区的 咨询和自己的努力,在解决这些问题的同时我们对Hive将SQL编译为MapReduce
  • 7个实例全面掌握Hadoop MapReduce

    万次阅读 多人点赞 2017-12-13 18:08:50
    本文旨在帮您快速了解 MapReduce 的工作机制和开发方法,解决以下几个问题: MapReduce 基本原理是什么? MapReduce 的执行过程是怎么样的? MapReduce 的核心流程细节 如何进行 MapReduce 程序...
  • MapReduce的写法

    千次阅读 2016-02-28 20:16:40
    开始聊mapreducemapreduce是hadoop的计算框架,我学hadoop是从hive开始入手,再到hdfs,当我学习hdfs时候,就感觉到hdfs和mapreduce关系的紧密。这个可能是我做技术研究的思路有关,我开始学习某一套技术总是想着...
  • MapReducemapreduce阶段添加依赖包办法 通常在MapReduce编程过程,大部分都是使用Hadoop jar的方式运行MapReduce程序,但是在开发mapreduce阶段中会需要引入外部的包,Hadoop集群环境并没有这些依赖包。 ...

空空如也

1 2 3 4 5 ... 20
收藏数 313,778
精华内容 125,511
关键字:

mapreduce