软件工程基础Hadoop生态系统刘驰AnEcosystemforCloudComputing课件.ppt

1、软件工程基础软件工程基础Hadoop生态系统生态系统刘刘 驰驰An Ecosystem for Cloud Computing2ProblempBatch (offline) processing of huge data set using commodity hardware is not enough for real-time applicationspStrong desire for linear scalabilitypNeed infrastructure to handle all mechanicspallow developers to focus on the proc

2、essing logic/algorithms3Explosive Data! StoragepNew York Stock Exchange: 1 TB data per daypFacebook: 100 billion photos, 1 PB (1000 TB)pInternet Archive: 2 PB data, growing by 20 TB per monthpCant put data on a SINGLE nodepStrong needs for distributed file systems5Java/Python/C interfaces6Commercial

3、 Hardware典型的典型的2层构架层构架 节点是普通的商业PC机 30-40 节点/Rack 顶层到Rack 带宽3-4Gbps Rack到节点带宽1Gbps7Who is (was) Using Hadoop?8Example: Facebook的的Hadoop集群集群p产品集群产品集群n4800个内核，600个机器，每个机器16GB2009年4月n8000个内核，1000个机器，每个机器32GB2009年7月n每个机器拥有4个1TB大小的SATA硬盘n两层网络结构，每个Rack有40个机器n整个集群大小为2PB，未来还会不断增加p测试集群测试集群800 个内核, 每个16GB 9A D

4、istributed File SystemSingle-Node ArchitectureMemoryDiskCPUMachine Learning, Statistics“Classical” Data Mining11Commodity ClusterspWeb data sets can be very large nTens to hundreds of TBpCannot mine on a single serverpStandard architecture emerging:nCluster of commodity Linux nodesnGigabit Ethernet

5、interconnectpHow to organize computations on this architecture?nMask issues such as hardware failure12Cluster ArchitectureMemDiskCPUMemDiskCPUSwitchEach rack contains 16-64 nodesMemDiskCPUMemDiskCPUSwitchSwitch1 Gbps between any pair of nodesin a rack2-10 Gbps backbone between racks13Stable storagep

6、First order problem: if nodes can fail, how can we store data persistently? pAnswer: Distributed File SystemnProvides global file namespacenGoogle GFS; Hadoop HDFS; Kosmix KFSpTypical usage patternnHuge files (100s of GB to TB)nData is rarely updated in placenReads and appends are common141516Nameno

7、de and DatanodespMaster/slave architecturep1 Namenode, a master server that manages the file system namespace and regulates access to files by clients.pmany DataNodes usually one per node in a cluster.nmanage storagenserves read, write requests, performs block creation, deletion, and replication upo

8、n instruction from Namenode.pHDFS exposes a file system namespace and allows user data to be stored in files.pA file is split into one or more blocks and set of blocks are stored in DataNodes.2022-6-2317Namespace2022-6-23pHierarchical file system with directories and filespCreate, remove, move, rena

9、me etc.pNamenode maintains the file systempAny meta information changes to the file system recorded by the Namenode.pAn application can specify the number of replicas of the file needed: replication factor of the file. This information is stored in the Namenode.18Data Replication2022-6-23pStore very

10、 large files across machines in a large cluster.pEach file is a sequence of blocks of same size.pBlocks are replicated 2-3 times.pBlock size and replicas are configurable per file.pNamenode receives a Heartbeat and a BlockReport from each DataNode in the cluster.pBlockReport contains all the blocks

11、on a Datanode.19Replica Placement2022-6-23pRack-aware: nGoal: improve reliability, availability and network bandwidth utilizationnResearch topicpNamenode determines the rack id for each DataNode.pReplicas are placed: n1 in a local rack, 1 on a different node in the local rack and 1 on a node in a di

12、fferent rack.n1/3 of the replica on a node, 2/3 on a rack and 1/3 distributed evenly across remaining racks.20HDFS: Data Node Distance21Replication PipeliningpWhen the client receives response from Namenode, it flushes its block in small pieces (4K) to the first replica, that in turn copies it to th

13、e next replica and so on.pThus data is pipelined from Datanode to the next.2022-6-2322Replica Selection 2022-6-23pReplica selection for READ operation: HDFS tries to minimize the bandwidth consumption and latency.pIf there is a replica on the Reader node then that is preferred.pHDFS cluster may span

14、 multiple data centers: replica in the local data center is preferred over the remote one.23Datanode2022-6-23pA Datanode stores data in files in its local file system.pDatanode has no knowledge about HDFS filesystempIt stores each block of HDFS data in a separate file.pDatanode does not create all f

15、iles in the same directory.pIt uses heuristics to determine optimal number of files per directory and creates directories appropriately: nResearch issue?pWhen the filesystem starts up it generates a list of all HDFS blocks and send this report to Namenode: Blockreport. 24HDFS: File Read25HDFS: File

16、Write26Communication Protocol2022-6-23pAll protocols are layered on top of the TCP/IP protocolpA client establishes a connection to a configurable TCP port on the Namenode machine. It talks ClientProtocol with the Namenode.pDatanodes talk to the Namenode using Datanode protocol.pRPC abstraction wrap

17、s both ClientProtocol and Datanode protocol.pNamenode is simply a server and never initiates a request; it only responds to RPC requests issued by DataNodes or clients. 27DataNode Failure and HeartbeatpDatanodes lose connectivity with Namenode.pNamenode detects this condition by the absence of a Hea

18、rtbeat message.pNamenode marks Datanodes without Hearbeat and does not send any IO requests to them.pAny data registered to the failed Datanode is not available to the HDFS.2022-6-2328Cluster RebalancingpHDFS architecture is compatible with data rebalancing schemes.pA scheme might move data from one

19、 Datanode to another if the free space on a Datanode falls below a certain threshold.pIn the event of a sudden high demand for a particular file, a scheme might dynamically create additional replicas and rebalance other data in the cluster.pThese types of data rebalancing are not yet implemented: re

20、search issue.2022-6-2329APIspHDFS provides Java API for application to use.pPython access is also used in many applications.pA C language wrapper for Java API is also available.pA HTTP browser can be used to browse the files of a HDFS instance.2022-6-2330FS Shell, Admin and Browser InterfacepHDFS or

21、ganizes its data in files and directories.pIt provides a command line interface called the FS shell that lets the user interact with data in the HDFS.pThe syntax of the commands is similar to bash and csh.pExample: to create a directory /foodir/bin/hadoop dfs mkdir /foodirpThere is also DFSAdmin int

22、erface availablepBrowser interface is also available to view the namespace.2022-6-2331A Distributed Computation Framework for Batch ProcessingWhat is Map/Reduce?pA Programming ModelpDecompose a processing job into Map and Reduce stagespDeveloper need to nprovide codes for Map and Reduce functionsnco

23、nfigure the jobnlet Hadoop handle the rest33MapReduce Model34Distributed Execution Overview UserProgramWorkerWorkerMasterWorkerWorkerWorkerforkforkforkassignmapassignreducereadlocalwriteremoteread,sortOutputFile 0OutputFile 1writeSplit 0Split 1Split 2Input Data35Example: Word CountpWe have a large f

24、ile of words, one word to a linepCount the number of appearances for each distinct wordpSample application: analyze web server logs to find popular URLs36Pseudo-Code: Word Countmap(key, value):/ key: document name; value: text of documentfor each word w in value:emit(w, 1)reduce(key, values):/ key:

25、a word; values: an iterator over countsresult = 0for each count v in values:result += vemit(key,result)37map(key=url, val=contents): For each word w in contents, emit (w, “1”)reduce(key=word, values=uniq_counts):Sum all “1”s in values listEmit result “(word, sum)”see bob runsee spot throwsee1bob1 ru

26、n1see 1spot 1throw1bob1 run1see 2spot 1throw138Word CountMapReducepInput: a set of key/value pairspUser supplies two functions:nmap(k,v) list(k1,v1)nreduce(k1, list(v1) v2p(k1,v1) is an intermediate key/value pairpOutput is the set of (k1,v2) pairs39What is MAP?p Map each data entry into a pair npEx

27、amplesnMap each log file entry into nMap day stock trading record into 40What is Shuffle/Merge phase?p Hadoop merges(shuffles) output of the MAP stage into:npExamplesnn 41What is Reduce?p Reduce entries produces by Hadoop merging processing into pairpExamplesnMap into nMap into 42Example uses: distr

28、ibuted grepdistributed sort web link-graph reversal term-vector / hostweb access log stats inverted index construction document clustering machine learning statistical machine translation . . .Widely ApplicableMapReduce Programs in Google Source Tree 43 100s/1000s of 2-CPU x86 machines, 2-4 GB of me

29、mory Limited bisection bandwidth Storage is on local IDE disks GFS: distributed file system manages data (SOSP03) Job scheduling system: jobs made up of tasks, scheduler assigns tasks to machines Implementation is a C+ library linked into user programs Implementation Overview44Implementation Overvie

30、wJob trackerTask trackerTask trackerTask trackerMaster NodeSlave node 1Slave node 2Slave node NWorkersuserWorkersWorkers45How It Works?46Data FlowpInput, final output are stored on HDFSnScheduler tries to schedule map tasks “close” to physical storage location of input datapIntermediate results are

31、stored on local FS of map and reduce workerspOutput is often input to another map reduce task47CoordinationpMaster data structuresnTask status: (idle, in-progress, completed)nIdle tasks get scheduled as workers become availablenWhen a map task completes, it sends the master the location and sizes of

32、 its R intermediate files, one for each reducernMaster pushes this info to reducerspMaster pings workers periodically to detect failures48FailurespMap worker failurenMap tasks completed or in-progress at worker are reset to idlenReduce workers are notified when task is rescheduled on another workerp

33、Reduce worker failurenOnly in-progress tasks are reset to idlepMaster failurenMapReduce task is aborted and client is notified49Execution 50Parallel Execution 51How Many Map and Reduce Jobs?pM map tasks, R reduce taskspRule of thumb:nM, R (# of nodes) in clusternOne DFS chunk per map is commonnImpro

34、ves dynamic load balancing and speeds recovery from worker failurepUsually R is smaller than M, because output is spread across R files52CombinerspOften a map task will produce many pairs of the form (k,v1), (k,v2), for the same key kne.g., popular words in Word CountpCan save network time by pre-ag

35、gregating at mapperncombine(k1, list(v1) v2nsame as reduce function53Partition FunctionpInputs to map tasks are created by contiguous splits of input filepFor reduce, we need to ensure that records with the same intermediate key end up at the same workerpSystem can use a default partition function e

36、.g., hash(key) mod RpSometimes useful to override ne.g., hash(hostname(URL) mod R ensures URLs from a host end up in the same output file54Execution SummarypHow is this distributed?1.Partition input key/value pairs into chunks, run map() tasks in parallel2.After all map()s are complete, consolidate

37、all emitted values for each unique emitted key3.Now partition space of output map keys, and run reduce() in parallelpIf map() or reduce() fails, re-execute!55Example: Trading Data ProcessingpInput: nHistorical Stock DatanRecords are CSV (comma separated values) text file nEach line : stock_symbol, l

38、ow_price, high_pricen1987-2009 data for all stocks one record per stock per daypOutput:nMaximum interday delta for each stock56Map Function: Part I57Map Function: Part II58Reduce Function59Running the Job : Part I60Running the Job: Part II61MapReduce : PageRankPageRank 是描述“random surfer”行为的模型C(t) 是t

39、的链接数, (1-d) 是阻尼系数(随机跳转)“random surfer”指的是随机的链接情况并不考虑具体的网页内容将pages rank平均分配给所有的链接网页阻尼系数用于描述“无聊”和随意的URLniiitCtPRddxPR1)()()1 ()(PageRank : 关键理解关键理解 p每次迭代的影响是有限的，第i+1次迭代仅仅依赖于第i次迭代p第i次迭代，每个节点的PageRank可以独立地计算PageRank 使用使用 MapReduce pM表示为系数矩阵pM的每行是分配给链接网页的PageRank值p这些值通过reduce聚集得到网页的PageRank值Map: 将PageRan

40、k 值分配给链接目标Reduce: 从多个源聚集PageRank值计算得到新的PageRank值不断迭代直至收敛Source of Image: Lin 2008阶段阶段 1: HTML处理处理 pMap任务输入(URL, page-content) 对进行map后输出(URL, (PRinit, list-of-urls)nPRinit是URL初始的PageRank值nlist-of-urls包含URL所指向的所有网页pReduce任务只是验证函数阶段阶段 2: PageRank分配分配 pReduce 任务输入(URL, url_list) 和许多(URL, val) 值n计算vals 并且计算d 来获得新的PR值n输出(URL, (new_rank, url_list)p非并行地检查是否收敛