基于Python的大数据Hadoop平台2-2、MapReduce

我不是女神ヾ 2024-02-20 11:02 62阅读 0赞

[资源下载地址][Link 1]：https://download.csdn.net/download/sheziqiong/88487676  
[资源下载地址][Link 1]：https://download.csdn.net/download/sheziqiong/88487676

## 大数据Hadoop平台2-2、MapReduce ##

## 介绍铺垫 ##

*  **开发语言**：Linux、python、java、c++、sql、scala
 *  **开发工具**：Linux、Hadoop、spark、tensorflow、pytorch
 *  **开发方式**：shell、vim、IDE(idea)
 *  **项目**：推荐系统----模板，融会贯通（检索、反作弊、预测）
 *  **重点**：架构思维，思考方式，解决方法。

在正式介绍MR之前，先铺垫一些Hadoop生态圈组件，如图所示，这些组件从下到上看，底层偏存储，上层多是计算框架，而zookeeper贯穿了整个生态。

### 1.Hadoop1.0生态圈 ###

![8c0ae5c487bbda593108056dbf5bd722.png][]

*  webserver—给用户提供服务
 *  Map Reduce:存储批量、离线任务，调度很多的机器节点共同处理数据
 *  Storm:实时计算
 *  Spark:批量（spark-core）、实时（stream）、机器学习（mllib）、统计和数据计算sql
 *  hive(SQL引擎)—本质：Map Reduce --封装简便工作—非专业编程人员使用
 *  mahout—本质：Map Reduce–>机器学习–封装算法库
 *  zookeeper—全局锁，分布式锁服务，协调员
 *  Hbase—分布式数据库，HDFS更上层的一个服务存在，本质还是存在HDFS上。
 *  HDFS—相当于新硬盘格式化成各种文件系统的格式，只不过是分布式文件系统，可以存储大量的数据。

### 2.常见业务 ###

#### 搜索 ####

*  **以前**：人们主动去搜索内容
 *  **现在**：手机网站主动给人推荐服务：天气提醒、外卖推荐、广告推荐等。现在的人每天拿着手机滑来滑去，精神需求已经不满足于内容索引，除了这些服务推荐，市面上比较火的抖音类软件也是此推荐道理。
 *  **名词**：
    
     *  分词的每一个结果叫token
     *  音乐–>物品（item）->itemid—网站量级base:几十万个
     *  用户–>人（user）->userid
 *  **大数据**–>用户多、物品多
    
     *  用户多：产生用户行为多（用户行为挖掘）
        
         *  用户行为：通过监控实时获取—个性化推荐（不同user推荐不同item）
     *  物品多：产生大量的元数据（Metadata）
        
         *  元数据：物品的属性Schema（描述数据的数据）

**个性化推荐**

![1993528fa2e5dea0c83b0a5ecc2ce965.png][]

**检索例子**

![6dc8a3ca2a10d54eab84e2eacb68a66f.png][]

**基于Map Reduce的建库系统（建库流）**

下载页面–>解析Parser–>倒排索引Inverter(网页包含字段)–>merger一下（归并）–>正式索引文件–>索引库里

![a7b8152a0e06040008a0367c6e7079ad.png][]

#### 广告 ####

*  搜索广告：搜索后得到的广告信息
 *  展示广告：推送
    
     *  点击、CPM展示曝光、CPS转化

#### 推荐 ####

![5be89e3b42bf87b82b9b2c5f613b7784.png][]

由此业务，大家应该可以看出不管是推荐还是搜索都是息息相关的。

以上简单给大家介绍了Hadoop平台的生态圈和一些应用业务，后面如果有时间会掺杂一些机器学习算法（NLP、推荐、分类、聚类、神经网络等）应用在里面做出简单的推荐系统。

重点：检索、推荐系统定位 架构层次理解

## 正文内容 ##

我们在上面看到Map Reduce只是生态圈中的计算框架，而这个生态圈是Hadoop1.0的，2.0后引入Yarn概念产生了很大改变，但总要有一个过度，这里就先介绍下HDFS1.0后再介绍基于1.0的MR是什么样的,再谈及后面的2.0.

上面的的文章我们已经介绍了早期对于大数据量（数据库分表问题Hash）和大流量请求（一致性哈希）的最简单的\#\#分而治之\#\#思想，那么面对大计算的情况怎么搞？

由此引出了Map Reduce这个计算框架。

我们通过上面的生态圈架构图很容易看出这个计算框架只是上层的一种封装，不管是输出还是输入源全部来自HDFS上，这里既然涉及到了HDFS，就简单介绍一下这个底层存储，后面会单独拉出来一篇文章介绍，其1.0和2.0有着很大区别。

我们通过前面的文章，已经搭建体验了一下Hadoop1.0的集群，主节点为master，从节点为slave1,slave2。

主是什么，就相当于部门的领导，从节点就是部门的员工，通常一个集群里面还有第三个角色client，是向master提交任务的，类似于公司的甲方，分配给master任务，然后master再分配给员工。

client读取数据时，通常并不知道所要读写的数据在那台机器上，这时就要请求master获取数据真实位置，并直接和slave建立某种联系，从而读写数据，但这里面还有着很多细节，会在后面的HDFS章节讲解，这里主要是为了铺垫存储关系。

好我们再继续介绍MR，这个要从以前的海量数据处理方式说起，原来处理数据的方式是集中式处理，也就是把海量数据不停的往同一个应用程序上挪，从而消耗了大量的时间成本在网络传输和磁盘的IO上，串行处理过大的数据量，性能低下。

此时需要改进一下计算流程和方法，我们把程序放到各个数据节点上，根据本地化原则只map处理本地的数据，并行处理完后得到多个中间结果，再交给reduce程序把各个地方的map结果merge到reduce节点，处理完后输出到datanode上。

map处理本地操作可以节省网络传输,在本地就可以把数据处理了.map程序适合于计算的本地化.我们的Reduce程序不能实现计算的本地化,因为是汇总map的输出,map的输出必然会分布在很多的机器上。而这里就引出了一个MR核心思想：分而治之。

这里举一个俗气例子：数钱：假如说有很多100元50元20元的钞票杂乱的放在桌子上，我应该如何利用这种思想去快速的解决它呢，这个时候一个人去一张张数和分类就会花费很多时间还有可能出错，那么我们就多找几个人一起工作，比如找103个人，前100个人每个人只需要规整自己面前的一堆钱，并且把这堆钱按照面值分堆，然后100个人工作结束，剩下三个人进来，100个人统一把三种面值发给三个人，100个人退出，而这三个人只需要数张数然后乘面值就好了。在这个例子里：100个人只做分类工作，3个人做合并工作。也就是先切分（map）再合并（reduce）。

上面这个例子还有一些细节：

*  **map**:就近原则，也就是说100个人只需要处理自己面前的钱
 *  **reduce**:如果3种面值分配给5个人，就必然会导致2个人闲置的情况

这些细节会在后面实践中有更深的体会。

![43dff9aa55de0cde21efe876f17ea83b.png][]

这里我画了一个图（比较丑），由这个图我们了解一下这个计算框架的执行流程：

首先真实数据源来自我们的HDFS的各个数据块上，这里先不管是在哪个机器上，我们的input输入源进来后先通过inputformat这个接口类的两个方法：split和recordReader对数据进行切分，切成一小块一小块的数据作为真正map这个算子真正要读入的数据源，而map拿到这个数据源就相当于100个人里面的一个人拿到自己的数据去做一些归并逻辑处理并输出在自己的节点上，之后按照key作一个hash为一个partition，partition是什么意思呢，这个就相当于我们在前面把每种面值分别分配给3个人一样的过程，partition0区域全是100面值，这个区域也就相当于给了一个reduce的输入去做一些逻辑处理然后对应的做一个输出。细心如你一定会发现一个inputspilt只对应一个map，一个partition区域对应了一个output这个细节，除了这个还有涉及这些方法的作用，shuffle的其他用法，后面会一点点介绍，这里先要有一个大概的执行流程概念。

![65f5acccd678e679d9eb82aa99e0a985.png][]

如图所示又多了一些细节，我们这个map其实就是个程序，拿到split数据做一些逻辑处理，而每个程序任务呢在计算机系统中必然占用维护着自己的进程空间或线程空间（这里的MR任务是进程，后面学spark为线程会有一个对比），map的结果必然先写在自己的内存空间中（buffer in memory），而占用的内存空间必然是有限的，通常默认是100M，很容易就写满了，而MR通常会满80%就把数据锁住，并清空内存把这些数据转储（dump）到磁盘（disk）中（溢写），而在转储的过程中不仅仅是简简单单的存储数据了，它会在这个过程中做一个shuffle（sort、partition等等），提前排序分好了很多小的区域（三种面值），随着split数据的不断涌入，我们磁盘上多出了很多这种小区域文件了，这里就需要一次归并排序，而归并排序的前提就是前面这些小文件已经是排好序的文件，而做完归并排序后的大文件也是已经排好序的了，这个归并排序merge也就是上面说过的partition分区的过程，最后每个map都得到了这样的三个区域的大文件，在这个图中和类比前面的100人，你会发现还有很多很多的map。接着再看reduce，我们的reduce设置了固定的个数，比如说我们图上的reduce只处理100元面值钞票，这时候他只是把每台机器节点上的100元这个区域的数据硬拷贝到自己这台机器上做逻辑处理，而这时候从每台机器过来的数据就是我们reduce上的输入源的一个个小数据，通过俩俩合并成为一个大的reduce，这个reduce再去做一个计数或者什么工作，并且输出。

我们把归并排序的过程细化，就是如图所示下面三个过程：

![d6eb12084563b5b25c57348541d49413.png][]

这里介绍数据存储格式：partition分区号，key(面值)，value（数量）

![4f087c9635083a67f10e74856bb5502a.png][]

这里介绍在map阶段本地做一个数据归并排序，本地化原则。

![f5ba75a2d315fc7118c862aebe4c8546.png][]

这里把每一个map及其节点的相同分区文件copy到reducer任务节点上并作merge为reduce程序进程的输入源。

在这里的开发中，我们只需要开发框架中的map和reduce逻辑代码就可以了，不需要去开发框架中的split、shuffle等等逻辑代码。map用来处理本机数据,在处理本地的数据时,不需要想我的数据存放在本机的什么位置,我要进行什么样的计算,计算结果我要放在本机的什么位置.这些东西都是由mapreduce框架给我们实现的,数据在哪,我们只需要知道hdfs就行了,数据处理之后的中间结果放在哪,这个也是mapreduce框架给我们做的,我们自己不需要管.reduce是把map输出的结果给汇总到一起,map输出的结果在哪,怎样传输到reduce中,我们开发人员也不需要管,我们只需要管数据汇总这一件事就可以了,处理之后的结果,只需要再写进hdfs中就可以了,我们的数据是放在hdfs中,hdfs决定把数据是放在哪个datanode上的,决定的权利不在于我们的处理,而是在于hdfs.到底放在哪个datanode上不需要我们去关心.-------后面HDFS映射关系会介绍位置。

那我们现在就开发一个小的word Count示例去套用上面的框架做分析：

#### map任务处理 ####

这里map收到的参数为文件的一行例如数据是：“hello lijie hello word spark scala java java java”，然后对上面的一行进行split(” “)切分，然后用context.write输出（hello ，1）（lijie ，1）（hello ，1）（word ，1）（spark ，1）（scala ，1）（java，1）（java，1）（java，1）也就是读取hdfs文件为内容,把内容中的每一行解析成一个个的键(key)值(value)对.文件总是有行的,键是字节的偏移量,值是每一行的内容,每一个键值对调用一次map函数.map函数处理输入的每一行.这里是for循环强制加1，key为word，value为数值。即自定义map函数,写自己的逻辑,对输入的key,value(把每一行解析出的key,value)处理,转换成新的key,value输出。

上面的输出（map的中间结果）会先写到一个缓冲区里面（环形缓冲区，默认100M），当写入百分之80的时候会对里面的数据进行dump，dump的过程会对里面的数据先进行分区（partition）然后排序，如果有combiner会进行局部的combiner，之后写入运行map程序的那台服务器的本地磁盘中，如果map一直执行，那么会每满百分之80又会执行上面的过程直到map执行完成。

上面的步骤走完之后产生了很多小的文件，然后会触发mr的文件合并，把多个文件进行合并，合并过程中又会进行排序和局部的combiner，如果定义的是2个reduce，那么最后每个map端就会生产2个分区文件，并且文件里面的内容会已经排序且局部combiner（前提是设置了combiner）

shuffle:把我们map中的数据分发到reduce中去的一个过程(Partion, Sort, Spill, Meger, Combiner, Copy, Memery, Disk……),基本都是在map这边的。

#### reduce任务处理 ####

map任务执行完成之后，reduce会从map端下载对应的文件，并且又会对下载过来的文件进行合并且排序并且会调用GroupComparator 对象（用来自定义哪些是一组的，mr程序默认key相同为同一组，但是可以自己定义GroupComparator，这样不同的key也可以进入同一个reduce方法进行处理，详情见：[http://blog.csdn.net/qq\_20641565/article/details/53491257][http_blog.csdn.net_qq_20641565_article_details_53491257]）

<group(textPair01，textPair02，textPair04),1 1 1>

执行reduce方法，并且执行reduce的逻辑，执行完成之后就会调用FileOutputFormat，这个同FileInputFormat一样，同样可以自定义，也是继承FileOutPutFormat返回一个RecordWriter,这里就不过多介绍，默认是TextFileOutputFormat写入到hdfs里面  
整个mapreduce的执行流程就完成了，如果FileOutputFormat和FileInputFormat是默认的话，那么数据流就是：HDFS -> 本地磁盘 ->HDFS，并且多次进行IO操作，所以mr的瓶颈在于他的IO操作，只适合进行离线计算。

![d3f2b02059b2bbd73398140a58506155.png][]

在Map Reduce计算框架中呢，有两个重要的进程：JobTracker和TaskTracker。

MapReduce是一种分布式计算模型,运行时不会在一台机器上运行,它是运行在很多的TaskTracker之上的。在我们的TaskTracker上面跑的是Map或者是Reduce Task任务。

*  master : JobTracker NameNode
 *  slave: TaskTracker DataNode
 *  进程：worker

JobTracker是主进程，TaskTracker是从进程。

JobTracker：用来接收客户的请求并且做任务调度，然后把这个任务分配到从节点。并兼任监控的工作，监控Task Tracker的健康状态，如果有一台发生故障，Job Tracker就会把任务分配给其他Task Tracker。JobTracker一直在等待JobClient提交作业 。

TaskTracker：是由JobTracker来派任务的，并且在本地开始进行工作，然后每隔3秒钟会发送心跳向主进程做一次工作的汇报和自身健康状况。

Slave主动向master拉生意。

![a593d768732fb5895505c07a82455920.png][]

这个图是什么意思呢？当工程师在本地机器（Job Client）上提交作业时，每提交一次任务就跟Job Tracker打一次交道，Job Tracker会返回一个id号和提交路径给Job Client，这个id就代表着我们的任务，Job Client就拿着这个id在HDFS上创建一个JobId名字的目录，这个目录里面存储的就是（Jar包、分片信息（代码）、作业配置信息），是从本地分发拷贝到HDFS上的，之后请求Job Tracker运行，Job Tracker拿到提交的需求（代码）后，这个老板就立刻制定方案（Job初始化），寻找空闲的员工机器Task Tracker工作，把HDFS上的需求数据进行分发给Task Tracker，与此同时Task Tracker在工作时也是不断地通过心跳机制汇报给Job Tracker当前的工作状态。

Client JVM —> JobTracker <—> TaskTracker:Child JVM

以上便是Map Reduce的任务执行流程。
    
    MR任务调度优先级默认FIFO：very_high、high、normal，low，very low
    
    这里我们通过一个python_wordcount的代码例子去理解：（这里的代码是Hadoop2.6.5的环境）后面我会发2.6.5的搭建笔记。
    run.sh脚本中，由于我们用python开发，就必须通过Hadoop streaming的方式提交作业：
    
    ````python
    HADOOP_CMD="/usr/local/src/hadoop-2.6.5/bin/hadoop"
    STREAM_JAR_PATH="/usr/local/src/hadoop-2.6.5/share/hadoop/tools/lib/hadoop-streaming-2.6.5.jar"#引入jar包提交
    
    INPUT_FILE_PATH_1="/The_Man_of_Property.txt" #HDFS的目录
    # INPUT_FILE_PATH_1="/a.txt"
    OUTPUT_PATH="/output"
    
    $HADOOP_CMD fs -rmr -skipTrash $OUTPUT_PATH #删除文件
    
    # Step 1.
    $HADOOP_CMD jar $STREAM_JAR_PATH \
        -input $INPUT_FILE_PATH_1 \
        -output $OUTPUT_PATH \
        -mapper "python map_new.py" \#如何执行代码
        -reducer "python red_new.py" \
        -file ./map_new.py \#分发代码到HDFS的各个节点
        -file ./red_new.py`

map代码:

import sys
    
    	for line in sys.stdin:
    		ss=line.strip().split(" ") #来了一行split以空格分隔
    		for word in ss:
    			print word.strip() #获取分割后的单词并strip分割制表符换行符之类的符号

head /文章 |python map.py

这时你就会看到文章的前十行的单词被分开成一个单词独占一行显示。  
好，我们再回头改一下输出：

print'\t'.join([word.strip(),'1'])

再次查看你会发现，我们把每个单词的后面都强制加了个1，相当于单词就是key，1就是value。

reduce代码:

import sys
    
    cur_word=None
    sum=0
    for line in sys.stdin:#标准输入map结果
    	ss=line.strip().split('\t')
    	if len(ss)!=2:
    		continue
    	word,cnt=ss
    	if cur_word==None:
    		cur_word=word
    	if cur_word!=word:
    		print'\t'.join([word,  str(sum)])
    		cur_word=word
    		sum=0
    	sum+=int(cnt)
    print'\t'.join([word,str(sum)])

单机调试成功后就可以run脚本把代码提交到集群了，不要忘了输入目录文件要先上传到集群中哦：hadoop fs -put 文件 /hdfspath

查看Hadoop目录命令：hadoop fs -ls /  
查看结果：hadoop fs -text /output/part-000000 |head

[资源下载地址][Link 1]：https://download.csdn.net/download/sheziqiong/88487676  
[资源下载地址][Link 1]：https://download.csdn.net/download/sheziqiong/88487676

[Link 1]: https://download.csdn.net/download/sheziqiong/88487676
[8c0ae5c487bbda593108056dbf5bd722.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/4e0ac136d9d84d8da0afa4328e84ad2c.png
[1993528fa2e5dea0c83b0a5ecc2ce965.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/ea1ea075fe2d4613bf0a003ead31ca58.png
[6dc8a3ca2a10d54eab84e2eacb68a66f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/ee818e5024f34885961a233bd1e1d25e.png
[a7b8152a0e06040008a0367c6e7079ad.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/86bb6ad350fa46ca9c56c4e11b78e785.png
[5be89e3b42bf87b82b9b2c5f613b7784.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/3613fd2a3c86448a90deac91ebf42899.png
[43dff9aa55de0cde21efe876f17ea83b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/b6ecb2543a1a42a0bf0eb71049cf56c4.png
[65f5acccd678e679d9eb82aa99e0a985.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/6d3ff2e418bd43ca9d3ff37d5507a713.png
[d6eb12084563b5b25c57348541d49413.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/117a71a38e5e4851b1e0d7928d5f7c2f.png
[4f087c9635083a67f10e74856bb5502a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/3da87da5a5cc418d9580863826fc8881.png
[f5ba75a2d315fc7118c862aebe4c8546.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/36de0dbcb9f24f2f9a4f9d22a8e9d829.png
[http_blog.csdn.net_qq_20641565_article_details_53491257]: http://blog.csdn.net/qq_20641565/article/details/53491257
[d3f2b02059b2bbd73398140a58506155.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/5bf4b7fa307f427ab1172f6181a0f57b.png
[a593d768732fb5895505c07a82455920.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/20/2f3fdd87199d404a973892af535fc0d7.png