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

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

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

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)

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

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)

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

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)<ts(B) \))

Google Map & Reduce

• Why?
  • Large-scale data processing
  • Computation cluster
• Features
  • Automatic parallelization/distribution
  • Fault tolerance
  • I/O scheduling
  • Monitoring/status update
• Google use case:
  • Web crawling (build index, compute PageRank)
• Idea:
  • Large amount input data (key-value-pairs)
  • Split processing into map- and reduce-task
• Software clones
  • Pig (Yahoo)
  • HBase
  • HIVE (Facebook)
  • Dryad (Microsoft)

Map operation

• Input: Key-value-pair
• Output: Set of intermediate key-value-pairs (could be another key than input)
• Execute on single node (easily parallelize)
• Central controller distributes job

Reduce operation

• Input: Intermediate key-values-pair
• Output: Final output (usually one tuple)
• Run on set of intermediate map results (with same key)
• Optimization: Sort/group map output data (equal keys are contiguous)
• Improvement: Combiner
  • Mini-reducer
  • Run in-memory after map-phase
  • Use group rare map keys into larger groups (e.g., others)
  • Goal: Reduce network traffic/overhead

Master server

• Assign map/reduce tasks to workers
• Assign map-tasks in batches (called split) (reduce communication/latency)
• Split implemented with hash function (distribution)
• Assign split to worker (start map-task for each key-value-pair)
• Check for node failure
  • Worker
    • Periodic heartbeats
    • Solution: Re-execution (re-execute complete split, re-execute reduce tasks)
    • Robust (no problem)
  • Master
    • No solution
• Check for task completion
• Route map-results to reduce-tasks

Example: Google PageRank

• Assign importance weight to websites
• Page linking to another page votes/gives credits
• \( PR(x) = \sum_{i=1}^N \frac{PR(t_i)}{deg^+(t_i)} \)
• Algorithm
  • Randomize PageRanks for each site
  • Iterate and distribute page ranks
  • Algorithm converges
• Using Map and Reduce:
  • Map: Distribute credit to link targets (intermediate key: URL, value: credit portion)
  • Reduce: Sum up credits
  • Repeat

Example: Pig

• Engine for executing programs for Hadoop
• Motivation
• Pig Latin: Language (extremely few lines)
• Data model
  • Atoms: Integer, string
  • Tuple: Sequence (any type)
  • Bag: Collection of tuples (not same type, not unique)
  • Map: String-Any-map

Example: Apache Spark

• Open-source processing framework
• Motivation:
  • MapReduce a lot of I/O (slow)
  • Keep data into main-memory
• 100x faster than disk
• Resilient distributed database (in-memory or disk)
• Operations
  • Transformations (sorting, grouping)
  • Actions (count, collect)
• More operations than Map and reduce
• More programming environments (Scala, R, Python, Java)