mapreduce_mapreduce原理 - CSDN
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-09-27 20:03:19
    一、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?这一篇就够了

    千次阅读 2020-09-04 20:21:04
    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介绍

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

    场景:比如有海量的文本文件,如订单,页面点击事件的记录,量特别大,单机版很难搞定。

    怎样解决海量数据的计算?

    求和: 1 + 5 +7 + 3 +4 +9 +3 + 5 +6
    在这里插入图片描述

    MapReduce产生背景

    如果让你统计日志里面的出现的某个URL的总次数,让你自己去写个单机版的程序,写个逻辑:无非就是读这个文件一行,然后把那个地方截取出来,截取出来之后,然后可以把它放到一个HashMap里面,用Map去重,看到一条新的URL ,就把它put进去,然后+1,如果下次看到再有就直接+1,没有就put进去,单机版的话逻辑是很好实现,但是数据量一大,你觉得单机版本还能搞定吗?

    首先2T的文件,你放在单机上可能存不下来,如果再他多一点呢?比如几千个文件,几十个T,单机存都存不下,那么存在哪里-------hdfs上。
    因为放在HDFS上可以放很多很多,比如说HDFS上有100个节点,每个节点上能耐挂载8T的硬盘,那就有800T,800T,你每个文件存3个副本的话,你至少也能存100多个T文件,耗费了大概6个T的空间,但是你一旦放到HDFS上就有一个问题:你的文件就会被切散了,被切三到很多的机器上,这个时候,你再对它们进行统计,这个时候,按照原来的逻辑,会不会出现问题?

    你的任何一个节点上存的是某个文件的某些块,假设你是在那台机器上去做统计的话,你统计到的永远是局部的数据,那你专门写一个客户端,我的程序运行在这个客户端上,我去读数据,读一点统计一点,到把整个文件都读完了,统计结果也就出来了,问题是那样的话,你的程序又变成了一个单机版的,那你的内存也就不够,因为你要第一点进来统计一点,是不是要保存一些中间数据,那你可能内存也不够了,而且你因为是一个单机版的程序,所以你的速度是不是也很慢,而且你还要不断的从网络里面去拿那些数据,也会很慢,所以这个时候呢?你专门写一个客户端去做统计,肯定是不合适。

    那你是不是应该把你的程序分发到集群的每一台DN上去做统计,也就是把运算往数据去移动,而不是把数据移动到运算,把我的运算逻辑移动到数据那端去,数据在哪里,我就在哪里运算,但是这也有一个问题,因为运算也变成了一个分布式的了,你的每一份运算结果都只是局部的结果,那么这个时候也存在问题:
    1.你的代码怎么实现分发到很多机器上去运行,这件事情谁帮你做,如果是要你自己写程序的话,你是不是得有个U盘拷你的jar包,一个计算一个计算的去拷,拷完之后再启动,每个机器都启动jar,等你启动最后一台的时候,前面那台已经运行完了,最后一台才刚开始启动,这个工作你用手动去做是不是不合适啊?所以当你把一个简单的逻辑,变成这种分布式运行的时候,你发现很多问题就来了:
    1)我的代码怎么分发,怎么配置启动环境,怎么启动起来,这个是不是得有一个庞大的系统去做,也就是说你应该额外开发这么一个叫做资源分发和Java启动程序这么一个配置 的系统,这个系统你会吗?写得出来吗?你要是写的话,是不是还得花很多时间,你还要写很多东西,因为你现在的Java不一定是擅长那个领域的 ,那这个耗费的代价就很大了
    2)那个数据,比如刚才那个日志数据,是放到HDFS上面去了,但是不代表HDFS上的每一台DN上面都有这一部分数据里面的内容,因为我们这个集群很大,你这个文件存进去的时候可能只占了其中的30台节点,其中某30台节点上有你这些文件,其他节点上个根本就没有你这些文件,那我们的代码,运行逻辑,最好是放到那30台上面去做统计,你放到其他的那些机器上,它可以运行,但是它的数据必须来源于网络,是不是效率会比较低,也就是说你的代码究竟分发到哪些机器上去运行,是不是也要一个策略的问题,那么这个策略是不是也有一定的算法,那么这个时候,你为了实现你那个简单的逻辑,就再去开发这样一个系统出来是不是也是很大的开发量,再考虑一个问题,假如刚才那两个工作你都做完了,你的代码真的成功的在那30台机器上跑起来了,跑起来之后,其中假设有一台机器宕机了,那么你统计的那一部分局部数据是不是也就没有了,那个局部结果也就没有了,那个局部结果没有,假设你有个汇总的结果还正确吗?没有意义了,也就是说你还得解决一个问题,就是你时时刻刻得却监控着你的程序运行情况,那个节点正常,哪个节点不正常,这个问题也是很复杂的,假设这个问题 你也解决了,还有一个问题:
    刚才你的逻辑只是统计出中间结果,这个时候是不是还得汇总啊,汇总就是意味着,你要在那30台节点之间结果里面才能汇总,要么把它们全部调到一台机器上就能进行汇总,但是你调到一台机器上汇总的话,你那一台机器的负载是不是会很高,对吧?假设我调到多态机器汇总,逻辑就变得复杂了,比如说,只要是你们那30台机器上统计,每台机器上的那个URL有多少条,你又把那个所有那个URL的数据全部汇总到某个汇总节点,假如有两个URL,哪个URL分发到那个节点上进行汇总,这个策略是不是也会变得很复杂了,那你还得去做个中间数据的调度系统,那也很麻烦。

    那这样的话,我们就发现,哪怕是一个很简单的东西,你也要把它变成一个分布式运行的程序,是不是面临很多很多其他的问题,跟我们逻辑无关的问题,往往这些问题要比解决那个逻辑要复杂得多,那么这些问题解决不是我们擅长的,我们大量的普通程序员还没达到那些程度,这么复杂的问题要写出来是不是很麻烦呢?你不能要求每个程序员都能达到那个功力把,我们不过就是写个简单的逻辑统计这个文本里面哪个URL出现的总次数,很简单的东西,所以呢,MapReduce才是我们这些普通程序员的福音。

    就是说当我们面临海量数据处理的时候,那个逻辑也许很简单,但是面临海量数据处理,要我们这个 逻辑代码变成分布式运行,就会变得很复杂,而那些很复杂的事情又不是我们关心的,我关心的只是那个逻辑,那这个时候,就有人把你不擅长的而且又必须解决的,而且跟你的逻辑关系又不大的那些东西全部给封装起来,那么这个时候,我们是不是就直接写逻辑了,就会MapReduce的框架和Yarn,这两个是不是做运算的,由这两个框架把我们刚才讲的那些东西全部封装起来,这个就是MapReduce产生的背景,就是这个问题。

    总结

    把我们很简单的运算逻辑很方便的扩展到海量数据的场景下分布式运算,所以MapReduce程序对我们程序员来说很简单,因为它把那些东西都给封装起来了,你只要写业务逻辑,写业务逻辑还不擅长吗?业务逻辑大部分就是处理文本,处理字符串,我们学的大部分逻辑里面,大部分都是在处理这个问题:处理文本、处理字符串、查询数据库是不是得到一些东西啊。查询一下数据库,处理一下字符串,输出结果,而这个逻辑本身 不用你太多的分布式细节 ,你只要把逻辑写出来就可以了,但是你写MapReduce的时候,必须要符合人家编程的规范,你不能你的写法写,他按他的写法写,每个人的写法都不一样,那MapReduce也没法给你去分发和运行,所以你也要符合他的规范,怎样才算符合规范呢?

    就说你的代码,你的任意一个逻辑实现都要分成这么两个步骤:
    1)Map
    2)Reduce

    比如说我们统计日志文件里面,相同URL出现的总次数
    如下图:
    在这里插入图片描述

    例子:

    在这里插入图片描述

    WCMapper.java

    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;
    /**
     * Description: Mapper<br/>
     * Copyright (c) , 2018, xlj <br/>
     * This program is protected by copyright laws. <br/>
     * Program Name:WCMapper.java <br/>
     * 
     * @version : 1.0
     * 
     * Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
     * 它的这个Mapper让你去定义四个泛型,为什么mapper里面需要四个泛型
     * 其实读文本文件的操作不用你来实现,框架已经帮你实现了,框架可以读这个文件
     * 然后每读一行,就会发给你这个map,让你去运行一次,所以它读一行是不是把数据传给你,
     * 
     * 那他传给map的时候,这个数据就意味着类型的一个协议,我以什么类型的数据给你,我是不是得事先定好啊
     * map接收的数据类型得和框架给他的数据类型一致,不然的话就会出现类型转换异常
     * 所以map里面得定数据类型,前面两个是map拿数据的类型,拿数据是以什么类型拿的,那么框架就是以这个类型传给你
     * 
     * 另外两个泛型是map的输出数据类型,即reduce也得有4个泛型,前面两个是reduce拿数据的泛型得和map输出的泛型类型一致
     * 剩下两个是reduce再输出的结果时的两个数据类型
     */
    /*
     * 4个泛型,前两个是指定mapper端输入数据的类型,为什么呢,mapper和reducer都一样
     * 拿数据,输出数据都是以<key,value>的形式进行的--那么key,value都分别有一个数据类型
     * KEYIN:输入的key的类型
     * VALUEIN:输入的value的类型
     * KEYOUT:输出的key的数据类型
     * VALUEOUT:输出的value的数据累心
     * map reduce的数据输入输出都是以key,value对封装的
     * 至于输入的key,value形式我们是不能控制的,是框架传给我们的,
     * 框架传给我们是什么类型,我们这里就写什么数据类型
     * 
     * 默认情况下框架传给我们的mapper的输入数据中,key是要处理的文本中一行的起始偏移量,
     * 因为我们的框架是读一行就调用一次我们的偏移量
     * 那么就把一行的起始偏移量作为key,这一行的内容作为value 
     * 
     * 那么输出端的数据类型是什么,由于我们输出的数<hello,1>
     * 那么它们的数据类型就显而易见了
     * 初步定义为:
     * Mapper<Long, String, String, int>
     * 但是不管是Long还是String,在MapReduce里面运行的时候,这个数据读到网络里面进行传递
     * 即各个节点之间会进行传递,那么要在网络里面传输,那么就意味着这个数据得序列化
     * Long、String对象,内存对象走网络都得序列化,Long、String,int序列化
     * 如果自己实现Serializable接口,那么附加的信息太多了
     * hadoop实现了自己的一套序列化机制
     * 所以就不要用Java里面的数据类型了,而是用它自己的封装一套数据类型
     * 这样就有助于提高效率,实现了自己的序列化接口
     * 在序列化传输的 时候走的就是自己的序列化方法来传递,少了很多负载信息,传递数据精简,
     * Long---LongWritable
     * String也有自己的封装-Text
     * int--IntWritable
     */
    public class WCMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    //	MapReduce框架每读一次数据,就会调用一次该方法
    	@Override
    	protected void map(LongWritable key, Text value, Context context)
    			throws IOException, InterruptedException {
    		//具体业务逻辑就写在这个方法体中,而且我们业务要处理的数据已经被框架传递进来,在方法参数中
    		//key--这一行数据的其实偏移量   value--这一行数据的文本内容
    		//1.先把单词拿出来,拿到一行
    		String line = value.toString();
    		//2.切分单词,这个是按照特定的分隔符 进行切分
    		String [] words = line.split(" ");
    		//3.把里面的单词发送出去
    		/*
    		 * 怎么发出去呢?我都不知道reduce在哪里运行
    		 * 其实呢,这个不用我们关心
    		 * 你只要把你的东西给那个工具就可以了
    		 * 剩下的就给那个框架去做
    		 * 那个工具在哪-----context
    		 * 它把那个工具放到那个context里面去了,即输出的工具
    		 * 所以你只要输出到context里面就行了
    		 * 剩下的具体往哪里走,是context的事情
    		 */
    		//遍历单词数组,输出为<K,V>形式 key是单词,value是1
    		for (String word : words) {
    			//记得把key和value继续封装起来,即下面
    			context.write(new Text(word), new IntWritable(1));
    		}
    		/*
    		 * map方法的执行频率:每读一行就调一次
    		 * 最后到reduce 的时候,应该是把某个单词里面所有的1都到,才能处理
    		 * 而且中间有一个缓存的过程,因为每个map的处理速度都不会完全一致
    		 * 等那个单词所有的1都到齐了才传给reduce
    		 */
    		//每一组key,value都全了,才会去调用一次reduce,reduce直接去处理valuelist
    		//接着就是写Reduce逻辑了
    		
    	}
    }
    
    

    WCReducer.java

    import java.io.IOException;
    
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    /**
     * Description: Reducer<br/>
     * Copyright (c) , 2018, xlj <br/>
     * This program is protected by copyright laws. <br/>
     * Program Name:WCReducer.java <br/>
     * 
     * @version : 1.0
     */
    /*
     * 类型记得要对应
     */
    public class WCReducer extends Reducer<Text, IntWritable, Text, Text> {
    	//map处理之后,value传过来的是一个value的集合 
    	//框架在map处理完成之后,将所有的KV对保存起来,进行分组,然后传递一个组,调用一次reduce
    	//相同的key在一个组
    	@Override
    	protected void reduce(Text key, Iterable<IntWritable> values, Context context)
    			throws IOException, InterruptedException {
    		//遍历valuelist,进行了累加
    		int count = 0;
    		for (IntWritable value : values) {
    			//get()方法就能拿到里面的值
    			count += value.get();
    		}
    		//输出一组(一个单词)的统计结果
    		//默认输出到HDFS的一个文件上面去,放在HDFS的某个目录下
    		context.write(key, new Text(count+""));
    		//但是还差一个描述类:用来描述整个逻辑
    	
    		/*
    		 * Map,Reducce都是个分散的,那集群运行的时候不知道运行哪些MapReduce
    		 * 
    		 * 处理业务逻辑的一个整体,叫做job
    		 * 我们就可以把那个job告诉那个集群,我们此次运行的是哪个job,
    		 * job里面用的哪个作为Mapper,哪个业务作为Reducer,我们得指定
    		 * 
    		 * 所以还得写一个类用来描述处理业务逻辑
    		 * 把一个特定的业务处理逻辑叫做一个job(作业),我们就可以把这个job告诉那个集群,
    		 * 
    		 */
    	}
    }
    
    

    WCRunner.java

    import java.io.IOException;
    
    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.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    
    /**
     * Description: 用来描述一个特定的作业<br/>
     * Copyright (c) , 2018, xlj <br/>
     * This program is protected by copyright laws. <br/>
     * Program Name:WCRunner.java <br/>
     * 
     * 该作业使用哪个类作为逻辑处理的map
     * 哪个作为reduce
     * 还可以指定该作业要处理的数据所在的路径
     * 还可以指定该作业输出的结果放到哪个路径
     * 
     * @version : 1.0
     */
    
    public class WCRunner {
    	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    		//首先要描述一个作业,这些信息是挺多的,哪个是map,哪个是reduce,输入输出路径在哪
    		//一般来说这么多信息,就可以把它封装在一个对象里面,那么这个对象呢就是 ----Job对象
    		Job job = Job.getInstance(new Configuration());
    		
    		//job用哪个类作为Mapper 指定输入输出数据类型是什么
    		job.setMapperClass(WCMapper.class);
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(IntWritable.class);
    		
    		//job用哪个类作为Reducer 指定数据输入输出类型是什么
    		job.setReducerClass(WCReducer.class);
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(Text.class);
    		
    		//指定原始数据存放在哪里
    		//参数1:里面是对哪个参数进行指定
    		//参数2:文件在哪个路径下,这个路径下的所有文件都会去读的
    		FileInputFormat.setInputPaths(job, new Path("input/data1"));
    		
    		//指定处理结果的数据存放路径
    		FileOutputFormat.setOutputPath(job, new Path("output1"));
    		
    		//提交
    		int isok =  job.waitForCompletion(true)?0:-1;
    		System.exit(isok);
    	}
    }
    

    在这里插入图片描述
    在这里插入图片描述

    在这里插入图片描述

    展开全文
  • 1.MapReduce概要介绍 2.MapReduce示例运行与解析 3.搭建Eclipse Hadoop开发环境 4.WordCount案例实践 5.Yarn原理及架构
  • MapReduce & Hadoop

    2009-09-23 22:56:39
    MapReduce &amp; Hadoop   2.2.1示例   考虑这样一个例子,在一个很大的文档集合中,计算每一个单词出现的次数。 可以使用下面的伪代码表示:   Map(String key, String value): //key:文档名 //...
  • java编写mapreduce并在hadoop中运行

    千次阅读 2018-03-06 16:22:59
    原文:...amp;utm_medium=referral1、mapreduce代码package test.mapreduce; import java.io.IOException; import java.util.StringTokenizer; import or...
  • 尚硅谷大数据技术之Hadoop(MapReduce) (作者:大海哥) 官网:www.atguigu.com 版本:V1.1 一 MapReduce概念 Mapreduce是一个分布式运算程序的编程框架,是用户开发“基于hadoop的数据分析应用”的核心框架; ...
  • 深入理解与应用Hadoop中的MapReduce

    万次阅读 2019-03-14 14:37:23
    现在大数据是越来越火了,而我自己研究这方面也很长时间了,今天就根据我自己的经验教会大家玩转MapReduce,下文中将MapReduce简写为MR。 本篇博客将结合实际案例来具体说明MR的每一个知识点。1、本篇博客核心...
  • 1. MapReduce基本编程模型和框架1.1 MapReduce抽象模型大数据计算的核心思想是:分而治之。如下图1所示。把大量的数据划分开来,分配给各个子任务来完成。再将结果合并到一起输出。 注:如果数据的耦合性很高,不能...
  • 一、MapReduce的概念 MapReduce是一种编程模型,用于大规模数据集(大于1TB)的并行运算。概念"Map(映射)"和"Reduce(归约)",和它们的主要思想,都是从函数式编程语言里借来的,还有从矢量...
  • 谷歌三大论文之一-MapReduce

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

    万次阅读 2018-06-29 07:09:53
  • MapReduce编程案例系列篇(01-15)

    万次阅读 多人点赞 2019-06-18 10:26:13
    由于本人最开始接触大数据工作,主要以写MapReduce程序为主,虽然现在有流行的言论称MapReduce这种运行很慢的分布式计算编程框架将要被各种内存计算框架取代。但是MapRedcue也会吸收很多流行的内存计算的各种优点,...
  • eclipse下开发mapreduce需要导入的jar包

    万次阅读 2016-06-16 15:18:08
    hadoop-2.7.2/share/hadoop/mapreduce下的所有jar包(子文件夹下的jar包不用) hadoop-2.7.2/share/hadoop/common下的hadoop-common-2.7.2.jar
  • Hadoop之MapReduce

    万次阅读 2014-03-17 20:58:21
    摘要:MapReduce是Hadoop的又一核心模块,从MapReduce是什么,MapReduce能做什么以及MapReduce的工作机制三方面认识MapReduce。 关键词:Hadoop MapReduce 分布式处理
  • MapReduce的优缺点

    千次阅读 2019-09-18 11:06:36
    3,容错性强:对于节点故障导致失败的作业,MapReduce计算框架会自动将作业安排到健康的节点进行,直到任务完成。 缺点 1,执行速度慢:普通的MapReduce作业几分钟完成,数据量大的可能几个小时甚至一天的时间。 2,...
  • MapReduce程序依赖的jar包

    千次阅读 2014-06-23 10:00:42
    难得想写个mapreduce程序,发现已经不记得需要添加那些jar包了,网上找了一会也没发现准确的答案。幸好对hadoop体系结构略知一二,迅速试出了写mapreduce程序需要的三个jar包。    不多不少,3个包足矣……贴出来...
  • MapReduce的Web界面

    千次阅读 2014-12-22 11:07:27
    Hadoop的Web界面用来浏览作业信息,对于跟踪作业运行进度,查找作业完成后的统计信息和日志非常有用。 浏览NameNode和JobTracker的网络接口,它们的地址默认为: NameNode : http://localhost:50070/ ...
  • 关于mapreduce 几个参数的解释

    千次阅读 2018-05-25 10:38:53
    一、mapreduce.map.java.opts、mapreduce.reduce.java.opts 以map任务为例,Container其实就是在执行一个脚本文件,而脚本文件中,会执行一个 Java 的子进程,这个子进程就是真正的 Map Task,mapreduce.map.java....
1 2 3 4 5 ... 20
收藏数 174,663
精华内容 69,865
关键字:

mapreduce