分布式数据库大作业.docx

分布式数据库大作业.docx

《分布式数据库大作业.docx》由会员分享，可在线阅读，更多相关《分布式数据库大作业.docx（17页珍藏版）》请在冰点文库上搜索。

分布式数据库大作业

分布式数据库大作业

HadoopMapReduce云计算模型研究

学院：

软件学院

专业：

软件工程

2014年01月02日

大连理工大学

HadoopMapReduce云计算模型研究

1.引言

Hadoop[1]是一个分布式系统基础架构，由Apache基金会开发。

用户可以在不了解分布式底层细节的情况下，开发分布式程序。

充分利用集群的威力高速运算和存储。

简单地说来，Hadoop是一个可以更容易开发和运行处理大规模数据的软件平台。

其包括两个部分：

HDFS[2]和MapReduce[3]。

DFS即HadoopDistributedFileSystem（Hadoop分布式文件系统），HDFS具有高容错性，并且可以被部署在低价的硬件设备之上。

HDFS很适合那些有大数据集的应用，并且提供了对数据读写的高吞吐率。

HDFS是一个master/slave的结构，就通常的部署来说，在master上只运行一个Namenode，而在每一个slave上运行一个Datanode。

HDFS支持传统的层次文件组织结构，同现有的一些文件系统在操作上很类似，比如你可以创建和删除一个文件，把一个文件从一个目录移到另一个目录，重命名等等操作。

Namenode管理着整个分布式文件系统，对文件系统的操作（如建立、删除文件和文件夹）都是通过Namenode来控制[4]。

MapReduce是由Google提出的一种并行分布式编程模型[5-7]。

用以进行大数据量的计算。

对于大数据量的计算，通常采用的处理手法就是并行计算。

至少现阶段而言，对许多开发人员来说，并行计算还是一个比较遥远的东西。

MapReduce就是一种简化并行计算的编程模型，它让那些没有多少并行计算经验的开发人员也可以开发并行应用。

MapReduce的名字源于这个模型中的两项核心操作：

Map和Reduce。

也许熟悉FunctionalProgramming（函数式编程）的人见到这两个词会倍感亲切。

简单的说来，Map是把一组数据一对一的映射为另外的一组数据，其映射的规则由一个函数来指定，比如对[1,2,3,4]进行乘2的映射就变成了[2,4,6,8]。

Reduce是对一组数据进行归约，这个归约的规则由一个函数指定，比如对[1,2,3,4]进行求和的归约得到结果是10，而对它进行求积的归约结果是24。

2.MapReduce的基本原理

MapReduce是云计算的核心技术之一，它为并行系统的数据处理提供了一个简单、优雅的解决方案。

其主要目的是为了大型集群的系统能在大数据集上进行并行工作，并用于大规模数据的并行运算。

DivideandConquer”是Mapreduce的核心思想[8]。

面对一个规模庞大的问题，要处理是以TB计的数据，Mapreduce采用“输入”------“分解”------“解决”------“聚合”------“输出结果”的基本过程。

在MapRedcue模型中用户只须指定一个map函数来处理一个输入的key/value对，产生中间结果key/value对集，再通过一个由用户指定的reduce函数来处理中间结果中具有相同key值的value。

适合用MapReduce来处理的数据集（或任务）有一个基本要求:

待处理的数据集可以分解成许多小的数据集，而且每一个小数据集都可以完全并行地进行处理[9-10]。

MapReduce通过把对数据集的大规模操作分发给网络上的每个节点实现可靠性；每个节点会周期性的把完成的工作和状态的更新报告回来。

如果一个节点保持沉默超过一个预设的时间间隔，主节点（类同GoogleFileSystem中的主服务器）记录下这个节点状态为死亡，并把分配给这个节点的数据发到别的节点。

每个操作使用命名文件的原子操作以确保不会发生并行线程间的冲突；当文件被改名的时候，系统可能会把他们复制到任务名以外的另一个名字上去。

（避免副作用）。

化简操作工作方式很类似，但是由于化简操作在并行能力较差，主节点会尽量把化简操作调度在一个节点上，或者离需要操作的数据尽可能近的节点上了；这个特性可以满足Google的需求，因为他们有足够的带宽，他们的内部网络没有那么多的机器。

在Google，MapReduce用在非常广泛的应用程序中，包括“分布grep，分布排序，web连接图反转，每台机器的词矢量，web访问日志分析，反向索引构建，文档聚类,机器学习，基于统计的机器翻译...”值得注意的是，MapReduce实现以后，它被用来重新生成Google的整个索引，并取代老的adhoc程序去更新索引。

　　MapReduce会生成大量的临时文件，为了提高效率，它利用Google文件系统来管理和访问这些文件。

3.框架的结构和工作流程

Mapreduce框架的主要程序分为三种即Master，Map和Reduce。

Master的主要功能有两个，任务的分割和任务的调度。

Master把输入文件切成许多个split，每个split文件一般为几十M。

Master同时还要调度任务监视各个mapworker和reduceworker的工作状态，以做出相应的安排。

Master还要监视各个子任务的完成进展情况。

Map的主要功能是读取经过切割split文件形成一个map任务，分析map任务，得到中间结构并且将同一类型的中间文件存放在同一个区域内等待特定的reduce程序读取。

另外不同的Reduce读取各个Map得到的特定的中间文件，将所有相同的中间文件整合成最后的输出文件。

其流程图如下图所示：

控制

分析key/value对

分区写入磁盘

读取

图1.基本控制流程图

说明：

1.单向箭头表示控制，双向箭头表示控制和反馈

2.某些操作中Mapworker硬盘上的key/value在被Reducerworker读取之前可以有combine操作，将相同key的value合并以减少读取次数

3.分散的输出文件也可以合并成一个输出文件而对于有些操作如求最大值则必须合并输出文件才能得到最终结果。

首先输入收据文件被Mapreduce库函数分割成M个split集。

用户定义的程序被拷贝到机群中，其中一个是master，其它的都是worker。

M个map任务和R个reduce任务将被分配。

Master负责调度任务和过程监视。

随时检测worker的工作状况，任务的完成进度。

Mapworker每完成一个子任务向master报告。

一个被分配了map任务的worker读取一个split集，该worker从这个split集中分析出key/value对，然后有map函数来处理这些key/value对并得到中间key/value对，这些key/value对将最终存放在mapworker的本地硬盘上。

每完成一个任务报告master。

中间key/value对被存在本地硬盘的R个不同的区域中，由于可能的key值很可能不止R个，故必须利用一个分割函数来划分中间文件，常用的是散列的方法（如hash（key）modR）。

保证key值相同的key/value对被存放同一区域中，并且将位置报告给master。

如果同一个key的中间文件多而小可以考虑用cmobine函数在本地进行合并。

当所有的split都被分析完成之后，reduceworker开始工作，每个reduce根据master的安排和信息指示利用机群的内部文件系统读取mapworker本地磁盘中特定位置的中间文件。

Reduce开始聚合中间文件，得到自己的输出文件。

在聚合的过程中由于有很多key值，一般将用到排序。

Reduceworker完成自己的工作后向master报告。

4.实验及测试

本文选取了三台Ubuntu虚拟机模拟分布式环境，搭建并完成了Hadoop自带的wordcount程序的测试。

搭建环境如下：

Ubuntu10.10、Hadoop1.0.4、jdk1.6.0_30

Namenode:

UB01（ip:

192.168.1.106）

Datanode:

UB02（ip:

192.168.1.104）UB03（ip:

192.168.1.107）

按如下步骤进行配置，此过程参考了文献[11]、[12]。

（1）配置NameNode和DataNode

修改每台机器的/etc/hosts（包括namenode和datanode）如下：

127.0.0.1localhost.localdomainlocalhost

192.168.1.106UB01

192.168.1.104UB02

192.168.1.107UB03

:

:

1localhost6.localdomain6localhost6

（2）在所有的机器上建立相同的用户

useraddpeter

passwd123456

输入密码后即可成功建立并设置peter用户和密码。

（3）SSH设置

由于启动hadoop时及任务执行过程中需要远程访问各主机，因此为了避免多次输入密码，应首先配置各主机的SSH使其能免输入的登录其他主机。

配置如下：

在所有机器的/home/peter目录下

ssh-keygen-trsa//一路回车到底

cd.ssh

然后在UB01上执行：

scpid_rsa.pubpeter@UB02:

/home/peter/.ssh/id_rsa.pub.UB01

scpid_rsa.pubpeter@UB03:

/home/peter/.ssh/id_rsa.pub.UB01

在UB02上执行：

scpid_rsa.pubpeter@UB01:

/home/peter/.ssh/id_rsa.pub.UB02

scpid_rsa.pubpeter@UB03:

/home/peter/.ssh/id_rsa.pub.UB02

在UB03上执行：

scpid_rsa.pubpeter@UB01:

/home/peter/.ssh/id_rsa.pub.UB03

scpid_rsa.pubpeter@UB02:

/home/peter/.ssh/id_rsa.pub.UB03

这样就把各自的密钥文件全部拷贝到每台机器上了，然后：

在UB01上执行：

catid_rsa.pubid_rsa_pub.UB02id_rsa.pub.UB03>>authorized_keys

在UB02上执行：

catid_rsa.pubid_rsa_pub.UB01id_rsa.pub.UB03>>authorized_keys

在UB03上执行：

catid_rsa.pubid_rsa_pub.UB02id_rsa.pub.UB01>>authorized_keys

这样向其他机器发起SSH连接，只有第一次登录时需要输入密码，以后就不需要。

（4）在所有机器上安装JDK

下载并解压jdk1.6.0_30到/usr/develop目录下，然后在/etc/profile最后追加下面内容：

JAVA_HOME=/usr/develop/jdk1.6.0_30

exportJRE_HOME=/usr/develop/jdk1.6.0_30/jre

exportCLASSPATH=.:

$JAVA_HOME/lib:

$JRE_HOME/lib:

$CLASSPATH

exportPATH=$JAVA_HOME/bin:

$JRE_HOME/bin:

$PATH

保存退出，然后执行source/etc/profile即可。

（5）在所有机器上安装配置Hadoop

可以首先在namenode上配置，配置后在分发到datanode上。

在这里，下载hadoop-1.0.4-bin.tar.gz，然后解压到/home/peter/hadoop/目录下。

接下来需要修改hadoop的conf文件夹下的配置信息：

修改hadoop-env.sh，加入如下语句，使得hadoop能够找到java的路径：

exportJAVA_HOME=/usr/develop/jdk1.6.0_29

修改core-site.xml，如下：

fs.default.name

hdfs:

//UB01:

9000

hadoop.tmp.dir

/home/grid/tmp

注意：

hadoop.tmp.dir是hadoop文件系统依赖的基础配置，很多路径都依赖它。

它默认的位置是在/tmp/{$user}下面，在local和hdfs都会建有相同的目录，但是在/tmp路径下的存储是不安全的，因为linux一次重启，文件就可能被删除。

导致namenode启动不起来。

修改hdfs-site.xml，如下:

dfs.replication

1

修改mapred-site.xml，如下：

mapred.job.tracker

192.168.1.106:

9001

masters里写入作为namenode节点机器的IP:

192.168.1.106

slaves里写入作为datanode节点的机器的IP:

192.168.1.104

192.168.1.107

到此，hadoop的有关配置已经完成，namenode端通过如下命令把配置好的hadoop发送到各个datanode处：

scp-rhadoop-1.0.4UB02:

/home/peter/hadoop

scp-rhadoop-1.0.4UB03:

/home/peter/hadoop

注意点：

关闭所有机器上的防火墙，不然会导致datanode起不来。

/etc/init.d/iptablesstop

在namenode端cd到hadoop文件夹下，格式化分布式文件系统：

bin/hadoopnamenode-format

下面接着在namenode端启动hadoop进程：

bin/start-all.sh

如果没有其它差错的话，hadoop可以正常启动，并能够看到如下结果：

在namenode端用jps命令查看启动情况，如下：

peter@UB01:

~/hadoop-1.0.4$jps

xxxxJps

xxxxNamenode

xxxxSecondarynamenode

xxxxJobTracker

在datanode端用jps查看启动情况，如下：

peter@UB02:

~/hadoop-1.0.4$jps

xxxxJps

xxxxDataNode

xxxxTaskTracker

然后可以通过如下地址来查看集群运行状况：

http:

//UB01:

50030

http:

//UB01:

50070

http:

//UB02:

50060

Hadoop自带了一个MapReduce测试程序，用于统计输入文件中各个单词出现的次数。

其核心代码及注释如下：

public class WordCount {  

  /** 

   * TokenizerMapper 继续自 Mapper 

   * 

   * [一个文件就一个map,两个文件就会有两个map] 

   * map[这里读入输入文件内容 以" \t\n\r\f" 进行分割，然后设置 word ==> one 的key/value对] 

   * 

   * @param Object  Input key Type:

   * @param Text    Input value Type:

   * @param Text    Output key Type:

   * @param IntWritable Output value Type:

   * 

   * Writable的主要特点是它使得Hadoop框架知道对一个Writable类型的对象怎样进行serialize以及deserialize. 

   * WritableComparable在Writable的基础上增加了compareT接口，使得Hadoop框架知道怎样对WritableComparable类型的对象进行排序。

   * 

   */  

  public static class TokenizerMapper  

       extends Mapper{  

    private final static IntWritable one = new IntWritable

（1）;  

    private Text word = new Text（）;  

    public void map（Object key, Text value, Context context  

                    ） throws IOException, InterruptedException {  

      StringTokenizer itr = new StringTokenizer（value.toString（））;  

      while （itr.hasMoreTokens（）） {  

        word.set（itr.nextToken（））;  

        context.write（word, one）;  

      }  

    }  

  }  

  /** 

   * IntSumReducer 继承自 Reducer 

   * 

   * [不管几个Map,都只有一个Reduce,这是一个汇总] 

   * reduce[循环所有的map值,把word ==> one 的key/value对进行汇总] 

   * 

   * 这里的key为Mapper设置的word[每一个key/value都会有一次reduce] 

   * 

   * 当循环结束后，最后的确context就是最后的结果. 

   * 

   * @author yangchunlong.tw 

   * 

   */  

  public static class IntSumReducer  

       extends Reducer {  

    private IntWritable result = new IntWritable（）;  

    public void reduce（Text key, Iterable values,  

                       Context context  

                       ） throws IOException, InterruptedException {  

      int sum = 0;  

      for （IntWritable val :

 values） {  

        sum += val.get（）;  

      }  

      result.set（sum）;  

      context.write（key, result）;  

    }  

  }  

  public static void main（String[] args） throws Exception {  

    Configuration conf = new Configuration（）;  

    String[] otherArgs = new GenericOptionsParser（conf, args）.getRemainingArgs（）;  

    /** 

     * 这里必须有输入/输出 

     */  

    if （otherArgs.length !

= 2） {  

      System.err.println（"Usage:

 wordcount  "）;  

      System.exit

（2）;  

    }  

    Job job = new Job（conf, "word count"）;  

    job.setJarByClass（WordCount.class）;//主类  

    job.setMapperClass（TokenizerMapper.class）;//mapper  

    job.setCombinerClass（IntSumReducer.class）;//作业合成类  

    job.setReducerClass（IntSumReducer.class）;//reducer  

    job.setOutputKeyClass（Text.class）;//设置作业输出数据的关键类  

    job.setOutputValueClass（IntWritable.class）;//设置作业输出值类  

    FileInputFormat.addInputPath（job, new Path（otherArgs[0]））;//文件输入  

    FileOutputFormat.setOutputPath（job, new Path（otherArgs[1]））;//文件输出  

    System.exit（job.waitForCompletion（true） ?

 0 :

 1）;//等待完成退出.  

  }  

}  

接下来，我们将hadoop目录下的conf目录作为输入，统计其下面所有文件中单词的个数。

步骤如下：

1.通过hadoop的命令在HDFS上创建input目录，命令如下：

bin/hadoopfs-mkdirinput

2.将conf目录下所有文件上传到HDFS的input目录，命令如下：

bin/hadoopfs–putconf/*input/

3.运行wordcount程序：

bin/hadoopjar