云计算与云数据管理

null 云计算与云数据管理 云计算与云数据管理陆嘉恒 中国人民大学 www.jiahenglu.net 《先进数据管理》前沿讲习班主要内容主要内容* 云计算概述 Google 云计算技术：GFS，Bigtable 和Mapreduce Yahoo云计算技术和Hadoop 云数据管理的挑战 人民大学新开的《分布式系统与云计算》课程人民大学新开的《分布式系统与云计算》课程* 分布式系统概述 分布式云计算技术综述 分布式云计算平台 分布式云计算程序开发 第一篇分布式系统概述第一篇分布式系统概述*第一章：分布式系统入门 第二章：客户-服务器端构架 第三章：分布式对象 第四章：公共对象请求代理结构 (CORBA) 第二篇 云计算综述 第二篇 云计算综述 *第五章：云计算入门 第六章：云服务 第七章：云相关技术比较 7.1网格计算和云计算 7.2 Utility计算（效用计算）和云计算 7.3并行和分布计算和云计算 7.4集群计算和云计算 第三篇 云计算平台第三篇 云计算平台*第八章：Google云平台的三大技术 第九章：Yahoo云平台的技术 第十章：Aneka 云平台的技术 第十一章：Greenplum云平台的技术 第十二章：Amazon dynamo云平台的技术 第四篇 云计算平台开发第四篇 云计算平台开发*第十三章：基于Hadoop系统开发 第十四章：基于HBase系统开发 第十五章：基于Google Apps系统开发 第十六章：基于MS Azure系统开发 第十七章：基于Amazon EC2系统开发Cloud computingCloud computingnullWhy we use cloud computing?Why we use cloud computing?Why we use cloud computing?Why we use cloud computing?Case 1: Write a file Save Computer down, file is lost Files are always stored in cloud, never lostWhy we use cloud computing?Why we use cloud computing?Case 2: Use IE --- download, install, use Use QQ --- download, install, use Use C++ --- download, install, use …… Get the serve from the cloudWhat is cloud and cloud computing?What is cloud and cloud computing?Cloud Demand resources or services over Internet scale and reliability of a data center. What is cloud and cloud computing?What is cloud and cloud computing? Cloud computing is a style of computing in which dynamically scalable and often virtualized resources are provided as a serve over the Internet. Users need not have knowledge of, expertise in, or control over the technology infrastructure in the "cloud" that supports them. Characteristics of cloud computingCharacteristics of cloud computingVirtual. software, databases, Web servers, operating systems, storage and networking as virtual servers. On demand. add and subtract processors, memory, network bandwidth, storage. nullIaaS Infrastructure as a ServicePaaS Platform as a ServiceSaaS Software as a ServiceTypes of cloud servicenullSoftware delivery model No hardware or software to manage Service delivered through a browser Customers use the service on demand Instant Scalability SaaSnullExamples Your current CRM package is not managing the load or you simply don’t want to host it in-house. Use a SaaS provider such as Salesforce.com Your email is hosted on an exchange server in your office and it is very slow. Outsource this using Hosted Exchange. SaaSnullPlatform delivery model Platforms are built upon Infrastructure, which is expensive Estimating demand is not a science! Platform management is not fun!PaaSnullExamples You need to host a large file (5Mb) on your website and make it available for 35,000 users for only two months duration. Use Cloud Front from Amazon. You want to start storage services on your network for a large number of files and you do not have the storage capacity…use Amazon S3. PaaSnullComputer infrastructure delivery model A platform virtualization environment Computing resources, such as storing and processing capacity. Virtualization taken a step furtherIaaSnullExamples You want to run a batch job but you don’t have the infrastructure necessary to run it in a timely manner. Use Amazon EC2. You want to host a website, but only for a few days. Use Flexiscale.IaaSCloud computing and other computing techniquesCloud computing and other computing techniquesThe 21st Century Vision Of ComputingThe 21st Century Vision Of ComputingLeonard Kleinrock , one of the chief scientists of the original Advanced Research Projects Agency Network (ARPANET) project which seeded the Internet, said: “ As of now, computer networks are still in their infancy, but as they grow up and become sophisticated, we will probably see the spread of ‘computer utilities’ which, like present electric and telephone utilities, will service individual homes and offices across the country.”The 21st Century Vision Of ComputingThe 21st Century Vision Of ComputingSun Microsystems co-founder Bill Joy He also indicated “It would take time until these markets to mature to generate this kind of value. Predicting now which companies will capture the value is impossible. Many of them have not even been created yet.”The 21st Century Vision Of ComputingThe 21st Century Vision Of ComputingDefinitionsDefinitionsutilityDefinitionsDefinitionsutilityUtility computing is the packaging of computing resources, such as computation and storage, as a metered service similar to a traditional public utilityDefinitionsDefinitionsutilityA computer cluster is a group of linked computers, working together closely so that in many respects they form a single computer. DefinitionsDefinitionsutilityGrid computing is the application of several computers to a single problem at the same time — usually to a scientific or technical problem that requires a great number of computer processing cycles or access to large amounts of data DefinitionsDefinitionsutilityCloud computing is a style of computing in which dynamically scalable and often virtualized resources are provided as a service over the Internet.Grid Computing & Cloud ComputingGrid Computing & Cloud Computingshare a lot commonality intention, architecture and technology Difference programming model, business model, compute model, applications, and Virtualization.Grid Computing & Cloud ComputingGrid Computing & Cloud Computingthe problems are mostly the same manage large facilities; define methods by which consumers discover, request and use resources provided by the central facilities; implement the often highly parallel computations that execute on those resources. Grid Computing & Cloud ComputingGrid Computing & Cloud ComputingVirtualization Grid do not rely on virtualization as much as Clouds do, each individual organization maintain full control of their resources Cloud an indispensable ingredient for almost every CloudnullAny question and any comments ?**Any question and any comments ?主要内容主要内容* 云计算概述 Google 云计算技术：GFS，Bigtable 和Mapreduce Yahoo云计算技术和Hadoop 云数据管理的挑战 Google Cloud computing techniquesGoogle Cloud computing techniquesThe Google File System The Google File System The Google File System (GFS)The Google File System (GFS)A scalable distributed file system for large distributed data intensive applications Multiple GFS clusters are currently deployed. The largest ones have: 1000+ storage nodes 300+ TeraBytes of disk storage heavily accessed by hundreds of clients on distinct machinesIntroductionIntroductionShares many same goals as previous distributed file systems performance, scalability, reliability, etc GFS design has been driven by four key observation of Google application workloads and technological environmentIntro: Observations 1Intro: Observations 11. Component failures are the norm constant monitoring, error detection, fault tolerance and automatic recovery are integral to the system 2. Huge files (by traditional standards) Multi GB files are common I/O operations and blocks sizes must be revisitedIntro: Observations 2Intro: Observations 23. Most files are mutated by appending new data This is the focus of performance optimization and atomicity guarantees 4. Co-designing the applications and APIs benefits overall system by increasing flexibilityThe DesignThe DesignCluster consists of a single master and multiple chunkservers and is accessed by multiple clientsThe MasterThe MasterMaintains all file system metadata. names space, access control info, file to chunk mappings, chunk (including replicas) location, etc. Periodically communicates with chunkservers in HeartBeat messages to give instructions and check stateThe MasterThe MasterHelps make sophisticated chunk placement and replication decision, using global knowledge For reading and writing, client contacts Master to get chunk locations, then deals directly with chunkservers Master is not a bottleneck for reads/writesChunkserversChunkserversFiles are broken into chunks. Each chunk has a immutable globally unique 64-bit chunk-handle. handle is assigned by the master at chunk creation Chunk size is 64 MB Each chunk is replicated on 3 (default) serversClientsClientsLinked to apps using the file system API. Communicates with master and chunkservers for reading and writing Master interactions only for metadata Chunkserver interactions for data Only caches metadata information Data is too large to cache.Chunk LocationsChunk LocationsMaster does not keep a persistent record of locations of chunks and replicas. Polls chunkservers at startup, and when new chunkservers join/leave for this. Stays up to date by controlling placement of new chunks and through HeartBeat messages (when monitoring chunkservers)Operation LogOperation LogRecord of all critical metadata changes Stored on Master and replicated on other machines Defines order of concurrent operations Also used to recover the file system stateSystem Interactions: Leases and Mutation OrderSystem Interactions: Leases and Mutation OrderLeases maintain a mutation order across all chunk replicas Master grants a lease to a replica, called the primary The primary choses the serial mutation order, and all replicas follow this order Minimizes management overhead for the MasterAtomic Record AppendAtomic Record AppendClient specifies the data to write; GFS chooses and returns the offset it writes to and appends the data to each replica at least once Heavily used by Google’s Distributed applications. No need for a distributed lock manager GFS choses the offset, not the clientAtomic Record Append: How?Atomic Record Append: How?Follows similar control flow as mutations Primary tells secondary replicas to append at the same offset as the primary If a replica append fails at any replica, it is retried by the client. So replicas of the same chunk may contain different data, including duplicates, whole or in part, of the same recordAtomic Record Append: How?Atomic Record Append: How?GFS does not guarantee that all replicas are bitwise identical. Only guarantees that data is written at least once in an atomic unit. Data must be written at the same offset for all chunk replicas for success to be reported.Detecting Stale ReplicasDetecting Stale ReplicasMaster has a chunk version number to distinguish up to date and stale replicas Increase version when granting a lease If a replica is not available, its version is not increased master detects stale replicas when a chunkservers report chunks and versions Remove stale replicas during garbage collectionGarbage collectionGarbage collectionWhen a client deletes a file, master logs it like other changes and changes filename to a hidden file. Master removes files hidden for longer than 3 days when scanning file system name space metadata is also erased During HeartBeat messages, the chunkservers send the master a subset of its chunks, and the master tells it which files have no metadata. Chunkserver removes these files on its ownFault Tolerance: High AvailabilityFault Tolerance: High AvailabilityFast recovery Master and chunkservers can restart in seconds Chunk Replication Master Replication “shadow” masters provide read-only access when primary master is down mutations not done until recorded on all master replicasFault Tolerance: Data IntegrityFault Tolerance: Data IntegrityChunkservers use checksums to detect corrupt data Since replicas are not bitwise identical, chunkservers maintain their own checksums For reads, chunkserver verifies checksum before sending chunk Update checksums during writesIntroduction to MapReduce Introduction to MapReduce MapReduce: InsightMapReduce: Insight ”Consider the problem of counting the number of occurrences of each word in a large collection of documents” How would you do it in parallel ? MapReduce Programming ModelMapReduce Programming Model Inspired from map and reduce operations commonly used in functional programming languages like Lisp. Users implement interface of two primary methods: 1. Map: (key1, val1) → (key2, val2) 2. Reduce: (key2, [val2]) → [val3] Map operationMap operation Map, a pure function, written by the user, takes an input key/value pair and produces a set of intermediate key/value pairs. e.g. (doc—id, doc-content) Draw an analogy to SQL, map can be visualized as group-by clause of an aggregate query. Reduce operationReduce operation On completion of map phase, all the intermediate values for a given output key are combined together into a list and given to a reducer. Can be visualized as aggregate function (e.g., average) that is computed over all the rows with the same group-by attribute. Pseudo-codePseudo-codemap(String input_key, String input_value): // input_key: document name // input_value: document contents for each word w in input_value: EmitIntermediate(w, "1"); reduce(String output_key, Iterator intermediate_values): // output_key: a word // output_values: a list of counts int result = 0; for each v in intermediate_values: result += ParseInt(v); Emit(AsString(result)); MapReduce: Execution overviewMapReduce: Execution overview MapReduce: ExampleMapReduce: Example MapReduce in Parallel: ExampleMapReduce in Parallel: Example MapReduce: Fault ToleranceMapReduce: Fault ToleranceHandled via re-execution of tasks. Task completion committed through master What happens if Mapper fails ? Re-execute completed + in-progress map tasks What happens if Reducer fails ? Re-execute in progress reduce tasks What happens if Master fails ? Potential trouble !! MapReduce: MapReduce: Walk through of One more Application null MapReduce : PageRank MapReduce : PageRank PageRank models the behavior of a “random surfer”. C(t) is the out-degree of t, and (1-d) is a damping factor (random jump) The “random surfer” keeps clicking on successive links at random not taking content into consideration. Distributes its pages rank equally among all pages it links to. The dampening factor takes the surfer “getting bored” and typing arbitrary URL. PageRank : Key InsightsPageRank : Key Insights Effects at each iteration is local. i+1th iteration depends only on ith iteration At iteration i, PageRank for individual nodes can be computed independently PageRank using MapReducePageRank using MapReduce Use Sparse matrix representation (M) Map each row of M to a list of PageRank “credit” to assign to out link neighbours. These prestige scores are reduced to a single PageRank value for a page by aggregating over them. PageRank using MapReduce PageRank using MapReduceSource of Image: Lin 2008 Phase 1: Process HTML Phase 1: Process HTML Map task takes (URL, page-content) pairs and maps them to (URL, (PRinit, list-of-urls)) PRinit is the “seed” PageRank for URL list-of-urls contains all pages pointed to by URL Reduce task is just the identity function Phase 2: PageRank Distribution Phase 2: PageRank Distribution Reduce task gets (URL, url_list) and many (URL, val) values Sum vals and fix up with d to get new PR Emit (URL, (new_rank, url_list)) Check for convergence using non parallel component MapReduce: Some More AppsMapReduce: Some More Apps Distributed Grep. Count of URL Access Frequency. Clustering (K-means) Graph Algorithms. Indexing Systems MapReduce Programs In Google Source Tree MapReduce: Extensions and similar appsMapReduce: Extensions and similar apps PIG (Yahoo) Hadoop (Apache) DryadLinq (Microsoft) Large Scale Systems Architecture using MapReduceLarge Scale Systems Architecture using MapReduceBigTable: A Distributed Storage System for Structured DataBigTable: A Distributed Storage System for Structured DataIntroductionIntroductionBigTable is a distributed storage system for managing structured data. Designed to scale to a very large size Petabytes of data across thousands of servers Used for many Google projects Web indexing, Personalized Search, Google Earth, Google Analytics, Google Finance, … Flexible, high-performance solution for all of Google’s productsMotivationMotivationLots of (semi-)structured data at Google URLs: Contents, crawl metadata, links, anchors, pagerank, … Per-user data: User preference settings, recent queries/search results, … Geographic locations: Physical entities (shops, restaurants, etc.), roads, satellite image data, user annotations, … Scale is large Billions of URLs, many versions/page (~20K/version) Hundreds of millions of users, thousands or q/sec 100TB+ of satellite image dataWhy not just use commercial DB?Why not just use commercial DB?Scale is too large for most commercial databases Even if it weren’t, cost would be very high Building internally means system can be applied across many projects for low incremental cost Low-level storage optimizations help performance significantly Much harder to do when running on top of a database layerGoalsGoalsWant asynchronous processes to be continuously updating different pieces of data Want access to most current data at any time Need to support: Very high read/write rates (millions of ops per second) Efficient scans over all or interesting subsets of data Efficient joins of large one-to-one and one-to-many datasets Often want to examine data changes over time E.g. Contents of a web page over multiple crawlsBigTableBigTableDistributed multi-level map Fault-tolerant, persistent Scalable Thousands of servers Terabytes of in-memory data Petabyte of disk-based data Millions of reads/writes per second, efficient scans Self-managing Servers can be added/removed dynamically Servers adjust to load imbalanceBuilding BlocksBuilding BlocksBuilding blocks: Google File System (GFS): Raw storage Scheduler: schedules jobs onto machines Lock service: distributed lock manager MapReduce: simplified large-scale data processing BigTable uses of building blocks: GFS: stores persistent data (SSTable file format for storage of data) Scheduler: schedules jobs involved in BigTable serving Lock service: master election, location bootstrapping Map Reduce: often used to read/write BigTable dataBasic Data ModelBasic Data ModelA BigTable is a sparse, distributed persistent multi-dimensional sorted map (row, column, timestamp) -> cell contents Good match for most Google applicationsWebTable ExampleWebTable Example Want to keep copy of a large collection of web pages and related information Use URLs as row keys Various aspects of web page as column names Store contents of web pages in the contents: column under the timestamps when they were fetched.RowsRows Name is an arbitrary string Access to data in a row is atomic Row creation is implicit upon storing data Rows ordered lexicographically Rows close together lexicographically usually on one or a small number of machinesRows (cont.)Rows (cont.)Reads of short row ranges are efficient and typically require communication with a small number of machines. Can exploit this property by selecting row keys so they get good locality for data access. Example: math.gatech.edu, math.uga.edu, phys.gatech.edu, phys.uga.edu VS edu.gatech.math, edu.gatech.phys, edu.uga.math, edu.uga.physColumnsColumns Columns have two-level name structure: family:optional_qualifier Column family Unit of access control Has associated type information Qualifier gives unbounded columns Additional levels of indexing, if desiredTimestampsTimestamps Used to store different versions of data in a cell New writes default to current time, but timestamps for writes can also be set explicitly by clients Lookup options: “Return most recent K values” “Return all values in timestamp range (or all values)” Column families can be marked w/ attributes: “Only retain most recent K values in a cell” “Keep values until they are older than K seconds”Implementation – Three Major ComponentsImplementation – Three Major ComponentsLibrary linked into every client One master server Responsible