大数据导论-第2章-大数据生态系统

大数据技术导论第二章大数据生态系统2.1

认识Hadoop2.2

HDFS2.3

MapReduce2.4

zookeeper1.3.2数据存储技术分布式数据库Bigtable分布式文件系统GFS分布式并行处理技术MapreduceHDFSHBaseMapReduce2.1认识hadoop第二章大数据生态系统1.1.1大数据产生历史必然——谷歌三驾马车2.1认识hadoop第二章大数据生态系统第二章大数据生态系统2.2

HDFS2.1认识Hadoop2.3

MapReduce2.4

zookeeper2.2.1HDFS体系结构2.2HDFS第二章大数据生态系统2.2.1HDFS体系结构2.2HDFS第二章大数据生态系统2.2.1HDFS体系结构2.2HDFS第二章大数据生态系统(1)HDFS=1个NameNode+1个SecondaryNameNode+多个DataNode；(2)NameNode称为名称节点、命名空间、主节点、元数据节点，主要存放元数据（meta）；账本。(3)SecondaryNameNode称为从元数据节点，是命名空间的冷备份（Namenode秘书），高可用（HA）解决方案，HDFS2.0采用另外一种机制。(4)DataNode称为数据节点，是存放数据的物理空间，以块（block）为基本单位；(5)NameNode和DataNode是主从结构；(6)块是HDFS操作最小单位，一般为128M；(7)元数据(meta)=文件目录结构信息(catalogue)+操作日志信息(log)，数据的描述信息(8)NameNode只存放catalogue，和SecondaryNameNode之间只相差3秒的catalogue；(9)DataNode按机架（rack）进行组织，图3.5有两个机架；(10)客户端只能同NameNode交互；2.2.1HDFS体系结构2.2HDFS第二章大数据生态系统(11)一个数据块通常要备份3份。第一份放到任务发起用户所在节点，如果请求是集群外发起，就随机选一不忙的节点；第二份放到与第一份不同的机架的节点上，第三份放到与第一份所在机架不同节点上，如图中标记黑色的块；(12)DataNode定时向NameNode发送状态信息（心跳，Heartbeats）,维护Block到本地文件系统（HDFSClient）的映射关系;(13)NameNode主要任务是：监控心跳、负载平衡、数据块备份的位置信息；(14)冷备份每3秒进行一次，SecondaryNameNode冷备份过程见图3.6；完成edits和fsimage同步，触发冷备份的条件：3600秒或edits大于64M(15)NameNode结构见图3.7；(16)HDFS白话描述：如果把HDFS比作一本书，Namenode存储的是书的目录，Datanode存储的就是书的正文内容，一章是一个文件，一节是一个块，目录称为元数据，目录指明的各章节页码称为映射，用户访问数据，首先要访问Namenode；(17)HDFS优点：采用抽象的块概念可以带来以下几个明显的好处：支持大规模文件存储；简化系统设计；适合数据备份；(18)HDFS缺点：块的大小远远大于普通文件系统，所以不适合存储小文件。2.2HDFS第二章大数据生态系统NameNode（主节点，其结构见图）2.2HDFS第二章大数据生态系统元数据细节2.2HDFS第二章大数据生态系统SecondaryNameNode（冷备份）冷备份过程如图所示。在冷备份合并过程中发生的操作记录在edits.new中。2.2HDFS第二章大数据生态系统2.2.2HDFS存储原理(1)读数据过程

类似图书馆借书。第1步：访问图书馆网站（Namenode），提出请求；第2步：返回请求结果(Metadata)；第3不：去图书馆借书（Datanode）；第4步：拿到想借的书（Read）2.2HDFS第二章大数据生态系统2.2.2HDFS存储原理(2)读数据算法描述2.2HDFS第二章大数据生态系统2.2.2HDFS存储原理(3)读数据代码实现importjava.io.BufferedReader;importjava.io.InputStreamReader;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.fs.FSDataInputStream;publicclassChapter3{publicstaticvoidmain(String[]args){try{Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");FileSystemfs=FileSystem.get(conf);Pathfile=newPath("test");FSDataInputStreamgetIt=fs.open(file);BufferedReaderd=newBufferedReader(newInputStreamReader(getIt));Stringcontent=d.readLine();//读取文件一行System.out.println(content);d.close();//关闭文件fs.close();//关闭hdfs}catch(Exceptione){e.printStackTrace();}

}}}2.2HDFS第二章大数据生态系统2.2.2HDFS存储原理(4)写数据策略

写数据包括增、删、改，与读数据过程不同。

第一个副本放置在上传文件的数据节点；如果是集群外提交，则随机挑选一台磁盘不太满、CPU不太忙的节点。

第二个副本放置在与第一个副本不同的机架的节点上。

第三个副本与第一个副本相同机架的其他节点上。

更多副本随机节点2.2HDFS第二章大数据生态系统2.2.2HDFS存储原理(5)写数据算法描述2.2HDFS第二章大数据生态系统2.2.2HDFS存储原理(6)写数据代码实现importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.FSDataOutputStream;importorg.apache.hadoop.fs.Path;publicclassChapter3{publicstaticvoidmain(String[]args){try{Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");FileSystemfs=FileSystem.get(conf);byte[]buff="Helloworld".getBytes();//要写入的内容Stringfilename="test";//要写入的文件名FSDataOutputStreamos=fs.create(newPath(filename));os.write(buff,0,buff.length);System.out.println("Create:"+filename);os.close();fs.close();}catch(Exceptione){e.printStackTrace();}2.2HDFS第二章大数据生态系统2.2.2HDFS存储原理(7)数据出错与恢复HDFS具有较高的容错性，可以兼容廉价的硬件，它把硬件出错看作一种常态，而不是异常，并设计了相应的机制检测数据错误和进行自动恢复，主要包括以下几种情形：名称节点出错、数据节点出错和数据出错。①名称节点出错

名称节点保存了所有的元数据信息，其中，最核心的两大数据结构是FsImage和Editlog，如果这两个文件发生损坏，那么整个HDFS实例将失效。因此，HDFS设置了备份机制，把这些核心文件同步复制到备份服务器SecondaryNameNode上。当名称节点出错时，就可以根据备份服务器SecondaryNameNode中的FsImage和Editlog数据进行恢复。②数据节点出错

每个数据节点会定期向名称节点发送“心跳”信息，向名称节点报告自己的状态

当数据节点发生故障，或者网络发生断网时，名称节点就无法收到来自一些数据节点的心跳信息，这时，这些数据节点就会被标记为“宕机”，节点上面的所有数据都会被标记为“不可读”，名称节点不会再给它们发送任何I/O请求。这时，有可能出现一种情形，即由于一些数据节点的不可用，会导致一些数据块的副本数量小于冗余因子。

名称节点会定期检查这种情况，一旦发现某个数据块的副本数量小于冗余因子，就会启动数据冗余复制，为它生成新的副本。HDFS和其它分布式文件系统的最大区别就是可以调整冗余数据的位置。

2.2HDFS第二章大数据生态系统2.2.2HDFS存储原理(7)数据出错与恢复③数据出错

网络传输和磁盘错误等因素，都会造成数据错误。

客户端在读取到数据后，会采用md5和sha1对数据块进行校验，以确定读取到正确的数据。

在文件被创建时，客户端就会对每一个文件块进行信息摘录，并把这些信息写入到同一个路径的隐藏文件里面。

当客户端读取文件的时候，会先读取该信息文件，然后，利用该信息文件对每个读取的数据块进行校验，如果校验出错，客户端就会请求到另外一个数据节点读取该文件块，并且向名称节点报告这个文件块有错误，名称节点会定期检查并且重新复制这个块。第二章大数据生态系统2.3

MapReduce2.1认识Hadoop2.4

zookeeper2.2

HDFSContents2.3.12.3.2实例分析：WordCount2.3.32.3.4MapReduce体系结构MapReduce工作流程MapReduce编程实践分布式并行编程MapReduce模型简介2.3.1MapReduce体系结构Map和Reduce函数问题：在MapReduce出现之前，已经有并行计算框架了，那么为什么Google还需要MapReduce？MapReduce与传统的并行计算框架相比有什么优势？一、分布式并行编程回顾传统并行计算框架MapReduce集群架构/容错性共享式(共享内存/共享存储)，容错性差非共享式，容错性好硬件/价格/扩展性刀片服务器、高速网、SAN，价格贵，扩展性差普通PC机，便宜，扩展性好编程/学习难度what-how，难what，简单适用场景实时、细粒度计算、计算密集型批处理、非实时、数据密集型分布式并行编程MapReduce模型简介2.3.2MapReduce体系结构Map和Reduce函数输入和输出都是键值对二、MapReduce模型简介2、一个例子二、MapReduce模型简介（1）MapReduce将复杂的、运行于大规模集群上的并行计算过程高度地抽象为两个函数：Map和Reduce（2）编程容易，不需要掌握分布式并行编程细节，也可以很容易把自己的程序运行在分布式系统上，完成海量数据的计算（3）MapReduce采用“分而治之”策略，一个存储在分布式文件系统中的大规模数据集，会被切分成许多独立的分片（split），这些分片可以被多个Map任务并行处理（4）MapReduce设计的一个理念就是“计算向数据靠拢”，而不是“数据向计算靠拢”，因为，移动数据需要大量的网络传输开销（5）MapReduce框架采用了Master/Slave架构，包括一个Master和若干个Slave。Master上运行JobTracker，Slave上运行TaskTracker

2、基本思想3、MapReduce的体系结构MapReduce体系结构主要由四个部分组成，分别是：Client、JobTracker、TaskTracker以及Task二、MapReduce模型简介1）Client用户编写的MapReduce程序通过Client提交到JobTracker端用户可通过Client提供的一些接口查看作业运行状态2）JobTrackerJobTracker负责资源监控和作业调度JobTracker监控所有TaskTracker与Job的健康状况，一旦发现失败，就将相应的任务转移到其他节点JobTracker会跟踪任务的执行进度、资源使用量等信息，并将这些信息告诉任务调度器（TaskScheduler），而调度器会在资源出现空闲时，选择合适的任务去使用这些资源二、MapReduce模型简介3）TaskTrackerTaskTracker会周期性地通过“心跳”将本节点上资源的使用情况和任务的运行进度汇报给JobTracker，同时接收JobTracker发送过来的命令并执行相应的操作（如启动新任务、杀死任务等）TaskTracker使用“slot”等量划分本节点上的资源量（CPU、内存等）。一个Task获取到一个slot后才有机会运行，而Hadoop调度器的作用就是将各个TaskTracker上的空闲slot分配给Task使用。slot分为Mapslot和Reduceslot两种，分别供MapTask和ReduceTask使用4）TaskTask分为MapTask和ReduceTask两种，均由TaskTracker启动二、MapReduce模型简介分布式并行编程MapReduce模型简介2.3.2MapReduce体系结构Map和Reduce函数程序WordCount输入一个包含大量单词的文本文件输出文件中每个单词及其出现次数（频数），并按照单词字母顺序排序，每个单词和其频数占一行，单词和频数之间有间隔输入输出HelloWorldHelloHadoopHelloMapReduceHadoop1Hello3MapReduce1World1三、Map和Reduce函数

一个WordCount执行过程的实例三、Map和Reduce函数三、Map和Reduce函数三、Map和Reduce函数函数输入输出说明Map<k1,v1>如：<行号,”abc”>List(<k2,v2>)如：<“a”,1><“b”,1><“c”,1>1.将小数据集进一步解析成一批<key,value>对，输入Map函数中进行处理2.每一个输入的<k1,v1>会输出一批<k2,v2>。<k2,v2>是计算的中间结果Reduce<k2,List(v2)>如：<“a”,<1,1,1>><k3,v3><“a”,3>输入的中间结果<k2,List(v2)>中的List(v2)表示是一批属于同一个k2的value2编写Map处理逻辑Map输入为<偏移量，一行文本>,范型<LongWritable,Text>Map输出为<单词，1>,范型<Text,IntWritable>2编写Map处理逻辑publicclassWordCountMapperextendsMapper<LongWritable,Text,Text,IntWritable>{

//对数据进行打散

@Override

protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{

//1.接入数据hellorebahellomimi

Stringline=value.toString();

//2.对数据进行切分

String[]words=line.split("");

//3.写出以<hello,1>

for(Stringw:words){

//写出reducer端

context.write(newText(w),newIntWritable(1));

}

}

}3编写Reduce处理逻辑Reduce的输入数据为<”China”,<1,1,1>>

，范型<Text,Iterable容器>

Reduce输出数据为<”China”,3>

，范型<Text,IntWritable>publicclassWordCountReducerextendsReducer<Text,IntWritable,Text,IntWritable>{//key->单词values->次数11111@Overrideprotectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{//1.记录出现的次数

intsum=0;for(IntWritablev:values){sum+=v.get();}//2.累加求和输出

context.write(key,newIntWritable(sum));}}7.6.3编写Reduce处理逻辑工作流程概述MapReduce各个执行阶段2.3.3MapReduce工作流程Shuffle过程详解应用程序执行过程3.1.2阿里云一、工作流程概述不同的Map任务之间不会进行通信不同的Reduce任务之间也不会发生任何信息交换用户不能显式地从一台机器向另一台机器发送消息所有的数据交换都是通过MapReduce框架自身去实现的工作流程概述MapReduce各个执行阶段2.3.3MapReduce工作流程Shuffle过程详解应用程序执行过程3.1.2阿里云二、MapReduce各个执行阶段HDFS以固定大小的block为基本单位存储数据，而对于MapReduce而言，其处理单位是split。split是一个逻辑概念，它只包含一些元数据信息，比如数据起始位置、数据长度、数据所在节点等。它的划分方法完全由用户自己决定。关于Split（分片）3.1.2阿里云二、MapReduce各个执行阶段Reduce任务的数量最优的Reduce任务个数取决于集群中可用的reduce任务槽(slot)的数目通常设置比reduce任务槽数目稍微小一些的Reduce任务个数（这样可以预留一些系统资源处理可能发生的错误）Map任务的数量Hadoop为每个split创建一个Map任务，split的多少决定了Map任务的数目。大多数情况下，理想的分片大小是一个HDFS块3.1.2阿里云二、MapReduce各个执行阶段工作流程概述MapReduce各个执行阶段2.3.2MapReduce工作流程Shuffle过程详解应用程序执行过程1.Shuffle过程简介3.1.2阿里云三、Shuffle过程详解2.Map端的Shuffle过程每个Map任务分配一个缓存MapReduce默认100MB缓存设置溢写比例0.8分区默认采用哈希函数排序是默认的操作排序后可以合并（Combine）合并不能改变最终结果在Map任务全部结束之前进行归并归并得到一个大的文件，放在本地磁盘文件归并时，如果溢写文件数量大于预定值（默认是3）则可以再次启动Combiner，少于3不需要JobTracker会一直监测Map任务的执行，并通知Reduce任务来领取数据合并（Combine）和归并（Merge）的区别：两个键值对<“a”,1>和<“a”,1>，如果合并，会得到<“a”,2>，如果归并，会得到<“a”,<1,1>>3.1.2阿里云三、Shuffle过程详解3.Reduce端的Shuffle过程Reduce端的Shuffle过程Reduce任务通过RPC向JobTracker询问Map任务是否已经完成，若完成，则领取数据Reduce领取数据先放入缓存，来自不同Map机器，先归并，再合并，写入磁盘多个溢写文件归并成一个或多个大文件，文件中的键值对是排序的当数据很少时，不需要溢写到磁盘，直接在缓存中归并，然后输出给Reduce3.1.2阿里云三、Shuffle过程详解工作流程概述MapReduce各个执行阶段2.3.2MapReduce工作流程Shuffle过程详解应用程序执行过程3.1.2阿里云四、MapReduce应用程序执行过程应用场景自然连接2.3.4MapReduce编程实践实战一、MapReduce应用场景MapReduce可以很好地应用于各种计算问题关系代数运算（选择、投影、并、交、差、连接）分组与聚合运算矩阵-向量乘法矩阵乘法应用场景自然连接2.3.4MapReduce编程实践实战假设有关系R(A，B)和S(B,C)，对二者进行自然连接操作使用Map过程，把来自R的每个元组<a,b>转换成一个键值对<b,<R,a>>，其中的键就是属性B的值。把关系R包含到值中，这样做使得我们可以在Reduce阶段，只把那些来自R的元组和来自S的元组进行匹配。类似地，使用Map过程，把来自S的每个元组<b,c>，转换成一个键值对<b,<S,c>>所有具有相同B值的元组被发送到同一个Reduce进程中，Reduce进程的任务是，把来自关系R和S的、具有相同属性B值的元组进行合并Reduce进程的输出则是连接后的元组<a,b,c>，输出被写到一个单独的输出文件中二、用MapReduce实现关系的自然连接用MapReduce实现关系的自然连接应用场景自然连接2.3.4MapReduce编程实践实战三、MapReduce实践1 任务要求2 编写Map处理逻辑3 编写Reduce处理逻辑4 编写main方法5 编译打包代码以及运行程序1任务要求文件A的内容如下：ChinaismymotherlandIloveChina文件B的内容如下：IamfromChina期望结果如右侧所示：I2is1China3my1love1am1from1motherland12编写Map处理逻辑Map输入类型为<key,value>期望的Map输出类型为<单词，1>2编写Map处理逻辑publicclassWordCountMapperextendsMapper<LongWritable,Text,Text,IntWritable>{

//对数据进行打散

@Override

protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{

//1.接入数据hellorebahellomimi

Stringline=value.toString();

//2.对数据进行切分

String[]words=line.split("");

//3.写出以<hello,1>

for(Stringw:words){

//写出reducer端

context.write(newText(w),newIntWritable(1));

}

}

}3编写Reduce处理逻辑在Reduce处理数据之前，Map的结果首先通过Shuffle阶段进行整理Reduce阶段的任务：对输入数字序列进行求和Reduce的输入数据为<key,Iterable容器>Reduce任务的输入数据：<”I”,<1,1>><”is”,1>……<”from”,1><”China”,<1,1,1>>

publicclassWordCountReducerextendsReducer<Text,IntWritable,Text,IntWritable>{//key->单词values->次数11111@Overrideprotectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{//1.记录出现的次数

intsum=0;for(IntWritablev:values){sum+=v.get();}//2.累加求和输出

context.write(key,newIntWritable(sum));}}7.6.3编写Reduce处理逻辑4编写main方法publicclassWordCountDriver{publicstaticvoidmain(String[]args)throwsIOException,ClassNotFoundException,InterruptedException{Configurationconf=newConfiguration();Jobjob=Job.getInstance(conf);//1.创建job任务

job.setJarByClass(WordCountDriver.class);//2.指定jar包位置

job.setMapperClass(WordCountMapper.class);//3.关联使用的Mapper类

job.setReducerClass(WordCountReducer.class);//4.关联使用的Reducer类

job.setMapOutputKeyClass(Text.class);//5.设置mapper阶段输出的数据类型

job.setMapOutputValueClass(IntWritable.class);

job.setOutputKeyClass(Text.class);//6.设置reducer阶段输出的数据类型

job.setOutputValueClass(IntWritable.class);

FileInputFormat.setInputPaths(job,newPath(args[0]));//7.设置数据输入的路径

FileOutputFormat.setOutputPath(job,newPath(args[1]));//8.设置数据输出的路径

boolean

rs=job.waitForCompletion(true);//9.提交任务

System.exit(rs?0:1);}}5编译打包代码以及运行程序实验步骤：使用java编译程序，生成.class文件将.class文件打包为jar包运行jar包（需要启动Hadoop）查看结果Hadoop2.x版本中的依赖jarHadoop2.x版本中jar不再集中在一个hadoop-core*.jar中，而是分成多个jar，如使用Hadoop2.6.0运行WordCount实例至少需要如下三个jar:$HADOOP_HOME/share/hadoop/common/hadoop-common-2.6.0.jar$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.6.0.jar$HADOOP_HOME/share/hadoop/common/lib/commons-cli-1.2.jar通过命令

hadoopclasspath

可以得到运行Hadoop程序所需的全部classpath信息将Hadoop的classhpath信息添加到CLASSPATH变量中，在~/.bashrc中增加如下几行：exportHADOOP_HOME=/usr/local/hadoopexportCLASSPATH=$($HADOOP_HOME/bin/hadoopclasspath):$CLASSPATH

执行

source~/.bashrc

使变量生效，接着就可以通过

javac

命令编译WordCount.java接着把.class文件打包成jar，才能在Hadoop中运行：运行程序：第二章大数据生态系统2.4zookeeper2.1认识Hadoop2.3

MapReduce2.2

HDFS什么是zookeeperzookeeper功能2.4.1Zookeeper概述zookeeper结构第二章大数据生态系统2.4.1zookeeper概述一、问题提出HDFSMapReduceHBASE如果HDFS、MapReduce……的主节点出故障怎么办？2.4.1zookeeper概述第二章大数据生态系统分布式集群协调器二、什么是zookeeper配置管理命名服务提供分布式同步集群管理第二章大数据生态系统2.4.1zookeeper概述三、zookeeper能做什么第二章大数据生态系统2.4.1zookeeper概述1、命名服务：一个分布式系统的各节点可能需要一个规范的、各节点的唯一的命名（例如节点名、CPU编号等），ZK可以实现这个的应用场景。2、配置管理：各个节点也会有一致的信息，例如每个节点的主配置信息。3、提供分布式同步：

为了管理方便，新加入的节点也需要快速同步这些信息。4、集群管理:

(1)集群状态监控和通知：分布式系统中的每个节点需要知道整个系统的状态、知道系统中每个节点的状态：当有新节点加入时它需要知道、当有节点出现故障时它需要知道、当有节点退出时它需要知道。ZK就是这样一个“通知工具”。(2)协调资源抢占（锁）：当分布式系统的多个节点试图同时抢占唯一资源时（例如同时写入一个文件），就需要对这个唯一资源的使用进行协调。这是ZK的“协调者”功能。(3)分派计算任务：如何协调1000个需要同时处理的任务到分布式系统的13个节点？如何保证执行失败的任务能被重新执行？如何在某个节点崩溃的情况下，接管其正在处理的任务？配置管理——订阅/发布订阅者（客户端）通过一定的方法告诉Zookeeper服务器，自己对哪个节点的数据感兴趣，那么在服务端数据发生变化时，就会