## The Cloud ### General - Goal: Data center usage - Problem: Classic approach not scalable enough for massive load (too expensive/inflexible) - Idea: Combine DDMBS and peer-to-peer - Nice-to-have features - Elasticity - Linear scalability - High availability - Low-end/cheap hardware - Flexible schemas - High fault-tolerance - Unified service interfaces #### Cloud computing - Definitions - Centralized, on-demand, pay-what-you-use provisioning of computation resources - Centrally unify computation resources - Self-service model (designed for failure) - Resources - Raw storage/CPU - Low-level services (runtime platform, abstracted storage space) - Software services (functionality provided by Cloud) - Full software functionality - Base problem: - IT infrastructure expensive (IT staff, server maintenance, hardware, software) - Load/utilization issue (required hardware, handle scaling, handle *spike*-loads) - Provide everything as a service - *Service*: - Provide strictly defined functionality - Service-level agreement (guarantees, e.g., latency, availability) ##### Types - Public cloud - *Traditional cloud computing* - Offered via Internet for general public - e.g., AWS, Google AppEndine - Features: Quick setup, scalability, flexibility, automated management - Target group: Start-ups - Problem: Security/compliance - Problem: Reliability (2010, Amazon S3 down for six hours) - Private cloud - Use behind cooperate firewall - Provided internally - Advantage - Fine grained control - More secure - Disadvantage - Hardware investment - Required for internal experts - Hybrid cloud (public + private) ##### Provided properties - Agility (resources are quickly available) - Costs (operational costs) - Independence (availability everywhere/for any device) - Multi-tenancy (large number of users, reduced costs, load distribution) - Reliability (durability, replication) - Scalability (provide new resources easily) - Low maintenance (maintain by specialized staff, low costs for user) - Transparency (costs directly visible, *pay-what-you-use*) ##### Software-as-a-service - Service/full application on demand - e.g., Word, Gmail - Administration/maintenance by Cloud provider ##### Platform-as-a-service - Resources/Software platform on demand - e.g., JRE, .NET - Customers run application on platform - No server maintenance necessary ##### Infrastructure-as-a-service - Physical assets/raw computation as service - e.g., virtual servers, Amazon EC2, VMWare cloud, Microsoft Azure - Provider can host multiple virtual servers - Use virtualization - Abstract physical hardware - Hypervisor (e.g., Xen, HyperV) - Task: Allocate available resources to VM - Dispatch/relocate VMs to machines - Problem virtual server: Limited size (limited through physical machine) - Solution: Use Pods ![](images/iaas.png) ###### Pods - Deploy applications in Pods - *Pod*: Group of machines that are tightly networked - Pod runs copies of given virtual servers with application and OS installed - Pods are responsible for disjoint part of workload - Scatter pods across availability zones (data centers, physically separated) - Important applications should have multiple copies per availability zone - Example: GMail - Load balancer assigns Pod for user session - Pods stored in GFS - Architectures (SAN, DAS, NAS) - SAN Pod - Servers have no own secondary storage - Advantages - Access to same data - Disadvantage - Expensive - Higher latency than DAS - DAS Pod - Server has own set of hard drives - Data from another server is difficult - Advantage - Cheap - Low latency - Disadvantage - No shared data access - Hard for live-migrate applications - Use abstractions: GFS, HDFS, Amazon SimpleDB ##### Applications - Amazon EC2 - Elastic compute cloud (public IaaS cloud) - Customer can rent virtual server (pay by CPU usage) - Virtual server can be configured - Google AppEngine - PaaS: Python or Java Runtime - Host application directly on AppEngine - Access resources to fixed maximum - Amazon SimpleDB - PaaS - Simple table-centric database engine - Similar to BigTable - SalesForce - SaaS customer-relationship-management (CRM) software - Features: - Customer database - Call center - Email - Knowledge base - Customer portal #### Vector clocks - Goal: - Detect conflicts using version numbers without central authority - Detect causality violations - Generate partially ordered labels - Formal - `$ N $` processes/nodes - `$ V_i\in \mathbb{N}^N $` - Knowledge of process `$ i $` about state of process `$ j $`: `$ V_{i,j} $` - Send message at `$ i $` - `$ V_{i,i} = V_{i,i} + 1 $` - Send complete timestamp with messages - Receive message `$ (m, V'_j) $` at `$ i $` - `$ V_{i,i} = V_{i,i} + 1 $` - `$ V_{i,j} = \max(V_{i,j}, V'_{i,j}) $` - Definition *causally-related* `$ A\rightarrow B $` - `$ \forall 1 \leq i \leq n : A_i \leq B_i \land \exists k : A_k < B_k $` - Definition *concurrent* `$ A \parallel B $` - `$ \neg (A\rightarrow B) \land \neg (B\rightarrow A) $` #### Big Data Trade-Offs - Drawbacks Dynamo - No guaranteed consistency - No complex data model (key-value store) - No transaction support - Drawbacks BigTable - No guaranteed consistency - No complex query language - No transaction support - Main problems: Replica consistency + transaction support - Common approaches for replica consistency - Synchronous Master-Slave replication - Asynchronous Master-Slave replication - Optimistic replication - Core problem: If a system crash happens in asynchronous replication, which state is correct? - Master-Slave: Master is correct - Optimistic: No global order - Requirement: Consensus - Solution: The Paxos Protocol ##### Synchronous Master-Slave replication - Master has primary copy - Mutations are pushed to slaves/replicas - Acknowledge mutation iff slave mutated successfully - Advantage - Replica always consistent - Disadvantage - Not partition-tolerant - No high availability - Bad scalability - Master Single Point of Failure ##### Asynchronous Master-Slave replication - Master pushes mutation to at least one slave/replica - Acknowledge mutation iff acknowledge by slave (wait for more to ensure consistency) - Push mutations to other slaves (e.g., BigTable) - Advantage - Used for efficient transactions - Disadvantage - Dirty read possible - Data loss/inconsistency possible - Complex repair after master failure (consensus protocol) ##### Optimistic replication - Homogeneous nodes - Any node can mutate data - Push to other replicas (e.g., Dynamo) - Acknowledge with Quorum - Advantage - High availability - Low latency - Disadvantage - Consistency problems - No transaction protocol (order unknown) #### Google hardware - Details revealed in 2007/2009 - Servers: - Custom-built - Estimation 2007: One million servers - Connections: - Connected to each other - Connected to internet hubs - Rack: 40-80 commodity-class x86 PCs - Outdated hardware - Unstable / cheap - Challenge: 100 % global uptime without data loss - 7 % of all internet traffic generated by Google - Annual operation costs (2007): 2.4 billion Dollar ### CAP-Theorem - Proposed by Brewer, proven by Gilbert/Lynch - Limitation of the design space of highly-available systems - Assumption: Highly-scalable distributed storage system with replication - Consistency - Consistency of replicas - CAP consistency similar to ACID atomicity - *All nodes see the same data* - Availability - Service is available and fully operational - *Always write, always read* - Partition-tolerance - Fault-tolerance: *No network failures (except total crash) causes incorrect responses* - Assumption: Short-term network partitions form frequently - Definition *Partition*: Set of connected nodes #### The theorem - Assumption: Highly-scalable distributed storage system with replication - Claim: It's only possible to archive at most two of the three properties - Degree is continuous (not binary) - Usually consistency is dropped - Drop only if partitions have formed - Use partition recover algorithms - Cases - Drop availability (C + P) 1. Synchronize all replicas (unscalable, e.g., DDBMS) 2. No replication (no high availability, bad latency, unscalable, e.g., CDBMS) - Drop consistency (A + P) - Keep one replica in each partition - Asynchronous messages for operations - *Eventual consistency* (versioning conflicts) - Example: Dynamo, BigTable, Cassandra #### BASE paradigm - Follows from CAP thereom - Motivation - Huge number of concurrent transactions - ACID problem: Lock contention thrashing - Sacrifice strict consistency - BASE transaction paradigm - Basic Availability (*always available*) - Soft-state (allow inconsistencies) - Eventual consistent - ACID-BASE is continuum - ACID: transactional consistency, scalability issue - BASE: eventual consistency, high scalability ### The PAXOS Protocol - Problem: - Reach consensus/data consistency for a value in distributed system that tolerate non-malicious failures - Paxos - Protocol family for solving consensus problem - Network of unreliable processors (non-malicious) - Byzantine Paxos: Malicious setting - Use case - Distributed log problem - Replicate transaction log - Use PAXOS to find consensus in log entries - Example: https://angus.nyc/2012/paxos-by-example/ - Applications using PAXOS - Google Megastore, 2011 - Google Chubby, 2006 - Petal (distributed storage), 1996 - Fragipani (scalable distributed FS), 1997 - Advantage - Scalable (against belief it'd be unscalable) #### Chosen value - Choose value iff majority of acceptors accept value in accept-phase - Abort decision - Conflicting prepare-messages from proposers - No majority/quorum - Finally: Restart with higher number - Learning the value 1. Inform learners directly 2. Inform proposer who broadcasts value to learners #### Properties - Guarantees: - Safety - Only consensus on value that is present in system - Choose one value only at a time - Only chosen values are learned (e.g., replicated) - Liveness - No deadlocking - Eventually choose a value - Each node learns about chosen value - Important properties - Proposal number is unique - Any two set of acceptors have at least one acceptor in common (ensure that no two proposals are accepted simultaneously) - Value in *accept-message* sent from proposer in phase two is the highest-numbered proposals of all responses in phase one #### Agents - Independent from nodes - Proposers - Gets mutation request - Starts voting (proposal) - Coordinator - Acceptors - Vote on item - Form multiple quorums (i.e., subset of acceptors) - Quorums share at least one member - Learners (replicate item) #### Algorithm 1. Prepare-phase - Proposer sends *prepare* requests to acceptors - Contain proposed value - Contain proposal number `$ n $` (monotonically increasing) - Acceptor `$ a $` receives *prepare* request - Proposal number `$ n $` - If `$ n $` is the highest number, promise to not accept any lower number - If `$ a $` has accepted a prepare-request with lower number, send old proposal number and value to proposer (only during phase 1) - If `$ n $` is lower than any previous accepted number, decline 2. Accept-phase - Proposer receives valid majority of promises - If any acceptor accepted a different proposal with smaller number, use the majority of returned proposals - If no previous proposals have been accepted, use own proposed value - Send accept message with chosen value to acceptors - Acceptor receives accept-message - Accept only of not promise was made to accept a higher-numbered proposal - Register value - Send accept-message to learners/proposer ### Amazon Dynamo #### Amazon Infrastructure - Service-oriented architecture - Each functionality encapsulated in a service (e.g., shopping cart, catalog, render) - Service level agreement - Description of each service (e.g., input, output) - Quality guarantees (e.g., maximum latency, response time) - 99.9 % of users must have an internal page render below 300 ms - Service: stateful/stateless - Amazon company - 6 million sales per day - Globally accessible/operating - Highly scalable - Services must be always available - Insight: RDBMS unsuitable #### General - Low-level distributed storage system - Requirements - Strict 99.9 % latency - *Always writeable* - User-perceived consistency (due to *sloppy quorum*) - Low cost (use commodity hardware) - Scalability (incrementally add nodes) - Tunable (tune trade-offs between costs, latency, consistency, conflict resolution strategy) - Design - Key-value-store (sufficient, no complex data model) - Nodes are non-malicious (trusted environment) - Nodes are altruistic (no personal agenda, e.g. not like peers in BitTorrent) - Dynamo cluster can be created by every service - Advantages - Highly available - Guaranteed low latencies - Incrementally scalable - Dynamic tuning - Disadvantages - No infinite scaling (fully meshed ring) - No real OLTP support (high load vs. inconsistencies) - Application must provide conflict resolution strategies #### Implementation - Similar to CHORD - Topology: Ring - Hash-based primary horizontal partitioning - Hash function - Hash nodes + data - MD5 (128 bit) - Assumptions - Nodes don't join/leave - Ring stays manageable (~ 10000 nodes) - Responsibility - Anti-clockwise arc - Load balancing - Virtual servers - Transfer completely - Scheme: Many-to-many (central controller) - Durability - Replication - Each item has a *coordinator* - Replicas on nodes from the *preference list* (e.g., the successor list) - Multiple nodes per virtual server - Advantages - Flexible storage - Balanced storage - Durable storage ##### Routing - Full finger table - Fully connected ring - Routing complexity `$ O(1) $` - Asynchronous propagation after write - Usually one Dynamo node per session - Dynamo node selection 1. Generic load balancer (simple) 2. Client (partition-aware library, more complex, less communication) ###### Request execution - Route to coordinator - Use first `$ N-1 $` healthy nodes from preference list - Quorums (`$ R+W>N $`) - `$ (3,2,2) $`: Consistent durable, interactive user state - `$ (n,1,n) $`: High performance read engine - `$ (1,1,1) $`: Distributed web cache (not durable, not consistent, very high performance) - Read operation - After `$ R-1 $` responses, forward to user (only send unique results) - Client handles conflict resolution - If versions are *causally-related*, Dynamo can resolve the conflict (*last-write-wins*) - Write operation - Coordinator writes locally - Buffering possible (trade-off durability and performance) - *Sloppy quorum* (consider only healthy nodes) - Successful write if `$ W-1 $` nodes confirm ##### Conflicts and consistency - Partition Mode - Branches may evolve - Use version numbers on items (vector clocks) - Causes for multiple version numbers: - No failure: Update in progress - Failure: Ring partitioned/massive failures - Conflict detection: - During client read/replication/anti-entropy protocol - Quorum - Anti-entropy protocol (detect inconsistent item) - Conflict resolution: - Use vector clocks - Start after conflict detection (handled by application or Dynamo if desired) - Strategies (application-dependent): - Last-write-wins (discard older branches) - Merging ###### Anti-entropy protocol - Conflict detection - Based on Merkle trees - Execution time: - Node re-joins the network after a failure, check if virtual server is consistent - Merkle tree - Used for integrity checking large files - Idea: - Leaf nodes are hashes of data blocks - Inner nodes are hashes of concatenation of leafs - Advantage: Parallel execution - Node has Merkle tree per virtual server - Exchange Merkle roots to check consistency - Inconsistency detected: Recursively check tree - Complexity - Merkle tree: `$ O_{worst}(N), O_{avg}(\log n) $` - Hash list: `$ O_{worst}(N), O_{avg}(N) $` ###### Versioning - Use vector clock for versioning - No global system time - Each node has - Logical clock (initially zero) - Global clock (small copy, clock from all reladed nodes) - Data item tagged with vector clock - Vector length - Dynamic (only the number of writing nodes) - Maximum: Ten entries (rarely more than ten are needed) ### Google File System - Distributed file system - Runs on standard POSIX-compliant linux file systems - Design and requirements - Run on commodity hardware - Handle large files (100 MB - 10 GB) - Optimize for append-operations - Optimize bandwidth (not latency) - Simple & extendible API - Guarantees - Atomic namespace mutation (using locks, log must be flushed to disk) - Relaxed consistency model for data mutations (chunks should be consistent and defined) #### Operations - Append operations - Most files: WORM - Map Reduce: Naturally append - GFS optimized for append-operations - Perform in parallel - Problem: Append-data too big for chunk - e.g., webcrawlers - Ratio append-to-write: `$ 8:1 $` - Write operations - Provide exact range - Problem: Collisions with other clients (locks could solve the problem) - Read operations - Sequential streams (large data, caching is useless) - Random reads (small data) #### Architecture & Implementation - GFS Cluster - Represents single file system - Single master server - Multiple chunk servers - Types of operations - Control flow - Data flow - Why? - Optimize bandwidth (assume nearest replica has highest bandwidth - Lighten load of master server - File System API - Flat file name space (no directories) - Not POSIX-compatible ![](images/gfs-arch.png) ##### Files - Split into fixed-sized chunks (like FS-blocks) - Replication: Three times - Metadata - Namespace - Access control information - File-chunk-mapping - Chunk location (with replicas) - Chunk version numbers ##### Chunks - Size: 64 MB - 64-bit unique global ID - 64 Byte metadata - Stored at chunk server - Definition *consistent chunk* - All clients see the same data - Definition *defined chunk* - All modification are visible - Advantages - Less requests - Less network overhead - Less metadata - Easy caching of metadata - Disadvantages - Possibility for hot spots - Increased risk for fragmentation ##### Master server - Single server - Important: Low load on master (no bottleneck) - Maintain file metadata - In main memory - Use soft-states - Use Heartbeat-messages to chunk servers to keep metadata up to date (append *leases*, see below) - Tasks - Load balancing - Migrate chunks to other chunk servers - Replication (three times) - Queries - Return chunk location - Return chunk locations following the requested - Chunk deletion - Delete metadata (chunk becomes stale) - Chunk becomes stale if chunk server fails an update - Garbage collection (physically delete stale chunks) - Chunk creation - Chunk allocation (Heuristic-based) 1. Servers with below-average disk space utilization 2. Limit the creation per time on chunk server 3. Spread across racks (physical) - Fault tolerance - Operation log - Log modifications to metadata (not frequent) - Log is replicated - Checkpoints for fast recovery (timeline) - Recovery: Shadow master + hot-standby (can be used to answer read queries) - Leases - Master grants *lease* for each chunk - Chunk server with lease is *primary chunk server* - Lease lasts one minute (append to heartbeat-message) ##### Chunk server - Multiple chunk servers per master - Contain chunk data - Primary chunk server - Receive *lease* from master for a certain chunk - Responsible for mutation operations - Request lease extension (in case of chunk mutation) #### Workflow for mutation operations 1. Client sends write request to master - Client knows chunk size - Client splits file into chunks - Message: (filename, chunk index) 2. Master sends chunk handle + replica locations (primary, secondary) 3. Client sends data to *best* replica - *Best replica*: e.g., clostest, highest bandwidth - Message: Payload + address range - Data stored in internal buffer (not written to disk) - Serially pipeline data to all replicas 4. Client sends write request to primary chunk server - Primary writes data atomically to disk - Primary determines serial order for data fragments 5. Primary sends write request to secondary replicas - Secondary replicas write in same order as primary - Inconsistent replicas for a short time 6. Secondary replicas notify primary replica 7. Primary notifies the client - Errors are reported to the client - Resolving error: Retry failed steps - Chunks may become inconsistent ![](images/gfs-write.png) ### Google BigTable - Challenges - Store (semi-)structured data (URLs, geographic data) - Data is large scale - High-performance NoSQL cluster-level structured storage system - Requirements - Continuous + asynchronous updates/processing - High read/write rates - Effective scanning - Scalability (incremental scaling) - Self-managing (auto load balancing) - Fault tolerance - Persistence (handled by GFS) - Cheap hardware - Environment - GFS (distributed fail-safe file system) - Chubby (distributed lock manager, bootstrapping) - MapReduce (programming model, parallel computing) - Drawbacks - Single table data model (no joins, no foreign keys, no constraints) - No complex query language (scan, direct key access) - Applications using BigTable - Google Analytics - Google Earth (row: geographic segment, column family: source) - Google Maps - Personalized Search (row: user ID, column family: action type) - Open-source BigTable clone: Hbase (Apache Hadoop project) #### Data Model - Focus on appends - Specialized to GFS - Rows are frequently added, not updated - Row update: Row append with new timestamp - Multi-dimensional sparse map - Cell: (row name, col name, timestamp) ##### Rows - Unique name (string) - Row access is atomic - Lexicographically ordered (similar rows are in nearby fragment) - Rows are not split ##### Columns - Two level name structure - Family name + qualifier name - Column family (same datatype, part of schema, ~100) - Individual columns possible (flexible data model) ##### Timestamps - 64 bit integer - Used for versioning - Update to cell creates new version with current timestamp - Versioning options: - Keep `$ n $` copies - Keep version with max. age `$ x $` #### API - Simple (Python, C++) - No complex query language - Features - Create/delete tables/column families - Modify metadata - Modify access control rights - Single-row transactions (no multi-row transactions) - Look up - Iterate over subset of rows #### Tablet - Fragment of a table - Contain contiguous range of rows (locality) - Tablet contains full rows (never split rows) - Disjoint - Size: 1 GB (multiple GFS chunks) - Base unit for load balancing/partitioning - Distributed tablet index (3-tier hierarchy) - Chubby file - Entrypoint (persistent, indestructible) - Point to tablet server with root tablet - Root tablet - Chubby file - Entrypoint - Never split - Link to metadata tablets - Store name range of metadata tablet - Metadata tablets - Index file - Store name range of data tablet - Links to data tablets - Data tablet (tablet containing the data) ![](images/bigtable-index.png) ##### Implementation of tablets - Assigned to exactly one tablet server - Components - Memtable - Log - Multiple Sorted String Table (SSTable) - Important aspects - Bloom filters - Minor compaction - Major compaction - Optimizations ![](images/tablet.png) ###### Bloom filter - Space-efficient data structure to probabilistically test set membership - Idea: - Bit array with length `$ N $` - Use `$ k $` different hash functions `$ h_1, ..., h_k $` - Insertion of `$ x $` - Hash each element with all `$ k $` hash function - Mark the buckets `$ h_1(x), ..., h_k(x) $` with `$ 1 $` - Test membership of `$ x $` - Hash each element - If one bucket contains is `$ 0 $`: No membership - If all buckets are `$ 1 $`: Possible membership ###### Memtable - All writes performed on memtable - Stored in main memory (volatile) - Sorted lexicographically ###### Log - Persistent append-only - Log for all write operations - Shared with all tablets of tablet server - Stored in GFS (persistent) ###### Sorted string table - Immutable ordered maps (easy traversing) - Contains tablet data (key-value pairs) - Keys: (row, column, timestamp) - Value: Payload - Stored in GFS (exactly one chunk) - Defined start/end key - SSTable ranges may overlap - No consistency problems (immutable, no locks) - SSTable can only be deleted completely (during *compaction*) - Structure - Multiple blocks of data (64 KB disk block, ordered map) - Single index block (mapping: key ranges to block number) - Metadata: Start key, end key - *Major compaction* - Perform periodically - Background maintenance - Compact the overlapping SSTables and memtable into non-overlapping SSTables - Delete records which are marked for deletion - Advantage: Increase performance (less merging, less accesses) ![](images/sstable.png) ###### Write operation - Goal: Ensure atomicity - Control flow 1. Check permissions (done with Chubby) 2. Generate log record 3. Content inserted into memtable (copy-on-write) - Case: Memtable reaches threshold (64 KB) - Freeze current memtable - Create new memtable - Serialize memtable to disk (*minor compaction*, overlapping SSTables) - Commit to log ###### Read operation - Control flow 1. Receive key/key range 2. Check permissions (done with Chubby) 3. Find blocks with matching range (using index block and key range) 4. Merge blocks into unified view - Efficiently - Use merge-sort - SSTable is sorted 5. Binary search on merged view - Problem: Access blocks even if queried rows are not present - Goal: Minimize physical access - Solution: Bloom filter (efficient in-memory technique to check if tuple is in SSTable) ###### Delete operation - Mark row as deleted - Rows are deleted in major compaction ###### Optimizations - Locality groups - Group related columns together (same/close SSTable) - Semantic knowledge (manually by developers) - Example: PageRank, WebCrawler - Use for compression (see below) - Compression - Most data easy-compressible (e.g., HTML) - Compress SSTable blocks individually - Locality group compression (e.g., two pages share components) - Two-pass frequent term compression #### Architecture - Runs on standard Google server nodes - Server runs multiple services - Application - Map and reduce worker - GFS chunk server instance - Bigtable server - Bigtable cluster/cell - Single master server - Multiple tablet server - GFS cluster (GFS master + Chunk server) - Chubby Lock Manager - Cluster Management server ![](images/bigtable-arch.png) ##### Master server - Never contacted by client - Tasks - Control/maintain tablet server - Assign/merge/split tablets - Add/remove servers - Garbage collection - Load balancing - Schema maintenance - Monitor unassigned tablets (check metadata tablet) - Detection of tablet server arrival - Monitor directory with tablet server lock files - Contact new servers - Detection of tablet server departure - Periodically obtain lock files of tablet servers - Lock is lost -> Node departed - Re-assign tablets - Restore membtable with logs (start with last minor compaction) - Delete tablet server lock file - Chubby file of master - Identification for master server - Periodically re-aquire master lock - If lock is granted to another master, master kills itself - If lock is freed, master has failed - Master server arrival - Load tablet assignment from root table - Check all tablet servers ##### Tablet server - Maintain multiple tablets (10-1000) - Load all tablets into memory - Clients know responsible tablet server - Failure: Each other node must take one tablet - Tablet server should be GFS primary replica - Tablet server runs GFS server - Tablet server requests the GFS lease for each tablet file - Advantage: Directly write to local disk - Node arrival - Register with Chubby - Create ID file - Request volatile lock for ID file - Periodically re-aquire lock - Master monitors directory with ID files - Node departure - ID file/lock expires - Noticed by master ### Google Megastore - History - Designed for Google AppEngine - Developer's requirements - Classic DB features (schema, indexes, joins, transactions) - Google infrastructure (high scalability, BigTable, GFS) - Idea: - High availability - Scalability - ACID semantics - Low latency - Keep *read-modify-write*-idiom - Architecture: - Based on BigTable/Chubby - Mix scalability with ACID - Synchronous replication using Paxos - Per-replica NoSQL datastore for log file - Megastore is geo-scale structured database - Features - Schemas (semi-relational tables) - ACID transactions (within entity groups, sloppy consistency spanning multiple entity groups) - Indexes - Log replication (Paxos) - Synchronous replication across data-centers (Paxos) - Drawbacks - Limited scalability - No powerful query language - Manual partitioning #### Schema ##### Structure - Schema has set of tables - Table has set of entities - Entity has set of strongly typed properties - Required / optional / multi-valued - Options for tables - Entity group root table - Entity group child table (single foreign key to root table) - *Entity group*: Root entity with related child tables (forms a fragment) - Support for multiple root tables (result in multiple entity group classes) ##### Mapping to BigTable - Entity mapped to BigTable row (primary keys for locality) - Entity groups are stored consecutively - BigTable column: Megastore table name + Megastore column name - `IN TABLE x`: Store Megastore table in BigTable table `x` ##### Index levels - Local index - For each entity group - Stored in group - Updated consistently/atomically) - Global index - Across all entity groups - No guarantee for consistency - Consistent update (?) - Requires latching (index locking, expensive) #### Transactions - Goal: - Geo-replication (across/within data centers) - Shard data by user location (locality) - Within data center all nodes are equal (better performance/availability) - Replication - Data center closest to user is the *master datacenter* - Transaction management in master datacenter - Push changes to replication server - Replicate each entity group independently - Synchronous across data centers - Use low-latency Paxos (Variation of Paxos) - Each entity group has own write-ahead log - ACID - Assumption: Entity group is mini-database (most operations performed on entity group, e.g., Users) - Serializable ACID within entity group (stronger than strict consistency) - Loose ACID across entity groups - Write-operation: Single-roundtrip in data center - Read-operation: Single-roundtrip pass in data center - Use MCVV-protocol - MultiVersion Concurrency Control - Increase concurrency (read old while work on newer versions) - Use timestamps (works because handling in one data center) - Write new version (timestamp) instead of updating - Used by BigTable, SAP HANA, PostgreSQL - Transactions: - Multiversion two-phase-commit - Multiversion timestamp ordering - Read-operation: Select *correct* value (depends on application) - Read consistency (Three levels) - Current (wait until all writes committed, using WAL) - Snapshot (read last committed value) - Inconsistent (read from nearest node, dirty read possible) - Write consistency - Get timestamp and log position of last committed transaction (current read) - Determine next available log position - Create log entry with the new value to write - Use Paxos for consensus of appending the new log entry (proposal number = timestamp) - Write to BigTable (learner) - Mutation within entity group: Sync (2PC) - Mutation across entity groups: Queues (async) or 2PC ### Google Spanner - Highly available global-scale distributed database - Architecture - Software stack - Directories - Data model - TrueTime - Features - Schema (semi-relational data model) - SQL-like query language - Global timestamps (TrueTime API) #### Architecture - Backend: Colossus (successor of GFS) - Two levels of servers: 1. *Universe*: Spanner deployment (*universemaster*) 2. *Zones*: Units of physical isolation - Cell: `$ (key, timestamp) \rightarrow string $` - *Spanserver*: - Part of Zone, similar to BigTable master - For each tablet: Paxos state machine - *Paxos Group*: Set of replicas #### Data model - Directory - Unit of data placement - Set of contiguous keys sharing a prefix - Same replication settings - Can be sharded - `INTERLEAVE IN PARENT`: Store foreign key members after the foreign key owner #### TrueTime - Distributed globally synchronized time - Based on GPS + atomic clocks - TrueTime API - `now(): TTInterval` (contains absolute time of invocation) - `after(t: TTInterval): boolean` - `before(t: TTInterval): boolean` - Time uncertainty below 6 ms - Datacenter - Majority: Time master with GPS antenna - Minority: Time master with atomic clock - Time daemon periodically (30 s) updates time 1. Query time master (nearby and far-away data centers) 2. Reach consensus - Time uncertainty `$ e $`: Increasing after a synchronization - Advantage: Globally meaningful commit timestamps (if A happened before B: `$ ts(A)