EDBT 2011 Tutorial

Divy Agrawal, Sudipto Das, and Amr El Abbadi

Department of Computer Science
University of California at Santa Barbara
EDBT 2011 Tutorial
EDBT 2011 Tutorial
 Delivering applications and services over the Internet:
 Software as a service

 Extended to:
 Infrastructure as a service: Amazon EC2
 Platform as a service: Google AppEngine, Microsoft Azure

 Utility Computing: pay-as-you-go computing

 Illusion of infinite resources
 No up-front cost
 Fine-grained billing (e.g. hourly)

EDBT 2011 Tutorial

EDBT 2011 Tutorial
 Experience with very large datacenters
 Unprecedented economies of scale
 Transfer of risk

 Technology factors
 Pervasive broadband Internet
 Maturity in Virtualization Technology

 Business factors
 Minimal capital expenditure
 Pay-as-you-go billing model
EDBT 2011 Tutorial
• Pay by use instead of provisioning for peak

Capacity
Resources

Resources
Demand Capacity

Demand
Time Time

Static data center Data center in the cloud

Unused resources
Slide Credits: Berkeley RAD Lab
EDBT 2011 Tutorial
• Risk of over-provisioning: underutilization

Capacity
Unused resources
Resources

Demand

Time

Static data center

Slide Credits: Berkeley RAD Lab

EDBT 2011 Tutorial
 • Heavy penalty for under-provisioning

Resources
Capacity
Resources

Demand
Capacity 2 3
1
Time (days)
Demand Lost revenue
1 2 3

Resources
Time (days)
Capacity

Demand
1 2 3
Time (days)
Lost users
EDBT 2011 Tutorial Slide Credits: Berkeley RAD Lab
 Unlike the earlier attempts:
 Distributed Computing
 Distributed Databases
 Grid Computing

 Cloud Computing is likely to persist:

 Organic growth: Google, Yahoo, Microsoft, and
Amazon
 Poised to be an integral aspect of National
Infrastructure in US and other countries

EDBT 2011 Tutorial

 Facebook Generation of Application Developers
 [Link]:
 Started with 50 servers on Amazon EC2
 Growth of 25,000 users/hour
 Needed to scale to 3,500 servers in 2 days
(RightScale@SantaBarbara)

 Many similar stories:

 RightScale
 Joyent
 …

EDBT 2011 Tutorial

EDBT 2011 Tutorial
EDBT 2011 Tutorial
 Data in the Cloud
 Platforms for Data Analysis
 Platforms for Update intensive workloads

 Data Platforms for Large Applications

 Multitenant Data Platforms

 Open Research Challenges

EDBT 2011 Tutorial
 Science
 Data bases from astronomy, genomics, environmental data,
transportation data, …

 Humanities and Social Sciences

 Scanned books, historical documents, social interactions data, …

 Business & Commerce

 Corporate sales, stock market transactions, census, airline traffic, …

 Entertainment
 Internet images, Hollywood movies, MP3 files, …

 Medicine
 MRI & CT scans, patient records, …

EDBT 2011 Tutorial

 Data capture and collection:
 Highly instrumented
environment
 Sensors and Smart Devices
 Network

 Data storage:
 Seagate 1 TB Barracuda @
$72.95 from [Link]
(73¢/GB)
EDBT 2011 Tutorial
  What can we do?
 Scientific breakthroughs
 Business process
efficiencies
 Realistic special effects
 Improve quality-of-life:
healthcare,
transportation,
environmental disasters,
daily life, …

 Could We Do More?
 YES: but need major
advances in our capability
to analyze this data

EDBT 2011 Tutorial

“Can we outsource our IT software and “We have terabytes of click-stream data –
hardware infrastructure?” what can we do with it?”

 Hosted Applications and services  Very large data repositories

 Pay-as-you-go model  Complex analysis
 Scalability, fault-tolerance,  Distributed and parallel data
elasticity, and self-manageability processing

EDBT 2011 Tutorial

 Data in the Cloud
 Platforms for Data Analysis
 Platforms for Update intensive workloads

 Data Platforms for Large Applications

 Multitenant Data Platforms

 Open Research Challenges

EDBT 2011 Tutorial
 Used to manage and control business
 Transactional Data: historical or point-in-time
 Optimized for inquiry rather than update
 Use of the system is loosely defined and can
be ad-hoc
 Used by managers and analysts to
understand the business and make
judgments

EDBT 2011 Tutorial

 Data capture at the user interaction level:
 in contrast to the client transaction level in the
Enterprise context

 As a consequence the amount of data

increases significantly

 Greater need to analyze such data to

understand user behaviors
EDBT 2011 Tutorial
 Scalability to large data volumes:
 Scan 100 TB on 1 node @ 50 MB/sec = 23 days
 Scan on 1000-node cluster = 33 minutes
 Divide-And-Conquer (i.e., data partitioning)

 Cost-efficiency:
 Commodity nodes (cheap, but unreliable)
 Commodity network
 Automatic fault-tolerance (fewer administrators)
 Easy to use (fewer programmers)
EDBT 2011 Tutorial
 Parallel DBMS technologies
 Proposed in the late eighties
 Matured over the last two decades
 Multi-billion dollar industry: Proprietary DBMS
Engines intended as Data Warehousing solutions
for very large enterprises
 Map Reduce
 pioneered by Google
 popularized by Yahoo! (Hadoop)
EDBT 2011 Tutorial
 Popularly used for more than two decades
 Research Projects: Gamma, Grace, …
 Commercial: Multi-billion dollar industry but
access to only a privileged few
 Relational Data Model
 Indexing
 Familiar SQL interface
 Advanced query optimization
 Well understood and well studied
EDBT 2011 Tutorial
 Overview:
 Data-parallel programming model
 An associated parallel and distributed
implementation for commodity clusters
 Pioneered by Google
 Processes 20 PB of data per day
 Popularized by open-source Hadoop project
 Used by Yahoo!, Facebook, Amazon, and the list is
growing …
EDBT 2011 Tutorial
 Raw Input: <key, value>

MAP

<K1, V1> <K2,V2> <K3,V3>

REDUCE

EDBT 2011 Tutorial

 Automatic Parallelization:
 Depending on the size of RAW INPUT DATA  instantiate
multiple MAP tasks
 Similarly, depending upon the number of intermediate
<key, value> partitions  instantiate multiple REDUCE
tasks
 Run-time:
 Data partitioning
 Task scheduling
 Handling machine failures
 Managing inter-machine communication
 Completely transparent to the
programmer/analyst/user
EDBT 2011 Tutorial
 Runs on large commodity clusters:
 1000s to 10,000s of machines
 Processes many terabytes of data
 Easy to use since run-time complexity hidden
from the users
 1000s of MR jobs/day at Google (circa 2004)
 100s of MR programs implemented (circa
2004)

EDBT 2011 Tutorial

 Special-purpose programs to process large
amounts of data: crawled documents, Web
Query Logs, etc.
 At Google and others (Yahoo!, Facebook):
 Inverted index
 Graph structure of the WEB documents
 Summaries of #pages/host, set of frequent
queries, etc.
 Ad Optimization
 Spam filtering
EDBT 2011 Tutorial
 Simple & Powerful
Programming Paradigm
For
Large-scale Data Analysis

Run-time System
For
Large-scale Parallelism &
Distribution

EDBT 2011 Tutorial

 MapReduce’s data-parallel programming model hides
complexity of distribution and fault tolerance

 Key philosophy:
 Make it scale, so you can throw hardware at problems
 Make it cheap, saving hardware, programmer and
administration costs (but requiring fault tolerance)

 Hive and Pig further simplify programming

 MapReduce is not suitable for all problems, but when it

works, it may save you a lot of time
EDBT 2011 Tutorial
 Parallel DBMS MapReduce

Schema Support  Not out of the box

Indexing  Not out of the box

Imperative
Declarative (C/C++, Java, …)
Programming Model
(SQL) Extensions through
Pig and Hive
Optimizations
(Compression, Query  Not out of the box
Optimization)
Flexibility Not out of the box 

Fault Tolerance Coarse grained techniques 

EDBT 2011 Tutorial

 Don’t need 1000 nodes to process petabytes:
 Parallel DBs do it in fewer than 100 nodes
 No support for schema:
 Sharing across multiple MR programs difficult
 No indexing:
 Wasteful access to unnecessary data
 Non-declarative programming model:
 Requires highly-skilled programmers
 No support for JOINs:
 Requires multiple MR phases for the analysis

EDBT 2011 Tutorial

 Web application data is inherently distributed on a
large number of sites:
 Funneling data to DB nodes is a failed strategy
 Distributed and parallel programs difficult to develop:
 Failures and dynamics in the cloud
 Indexing:
 Sequential Disk access 10 times faster than random
access.
 Not clear if indexing is the right strategy.
 Complex queries:
 DB community needs to JOIN hands with MR

EDBT 2011 Tutorial

 Asynchronous Views over Cloud Data
 Yahoo! Research (SIGMOD’2009)

 DataPath: Data-centric Analytic Engine

 U Florida (Dobra) & Rice (Jermaine) (SIGMOD’2010)

 MapReduce innovations:
 MapReduce Online (UCBerkeley)
 HadoopDB (Yale)
 Multi-way Join in MapReduce (Afrati&Ullman: EDBT’2010)
 Hadoop++ (Dittrich et al.: VLDB’2010) and others
EDBT 2011 Tutorial
 Data-stores for Web applications & analytics:
 PNUTS, BigTable, Dynamo, …
 Massive scale:
 Scalability, Elasticity, Fault-tolerance, &
Performance
Weak consistency ACID

Simple queries Not-so-simple queries SQL

Views over Key-value Stores

EDBT 2011 Tutorial
 Simple Not-so-simple SQL
0

 Primary  Secondary access

access
▪ Point lookups  Joins
▪ Range scans  Group-by aggregates

EDBT 2011 Tutorial

 Reviews(review-id,user,item,text)

1 Dave TV … 4 Alex DVD … 7 Dave GPS …

2 Jack GPS … 5 Jack TV … 8 Tim TV …

Reviews for TV ByItem(item,review-id,user,text)

TV 1 Dave …
DVD 4 Alex … GPS 2 Jack …
TV 5 Jack
GPS 7 Dave …
TV 8 Tim …
View records are replicas
with a different primary key ByItem is a remote view
EDBT 2011 Tutorial
 Clients 

API
Query Routers
a. disk write in log Log Managers
b. cache write in storage
c.
 return to user f
d. flush to disk a
e. remote view maintenance
f. view log message(s)
g. view update(s) in storage

db
g e g
Storage Servers Remote Maintainer

EDBT 2011 Tutorial

 An architectural hybrid of MapReduce and
DBMS technologies
 Use Fault-tolerance and Scale of MapReduce
framework like Hadoop
 Leverage advanced data processing
techniques of an RDBMS
 Expose a declarative interface to the user
 Goal: Leverage from the best of both worlds

EDBT 2011 Tutorial

EDBT 2011 Tutorial
EDBT 2011 Tutorial
 MapReduce works with a single data source (table):
 <key K, value V>

 How to use the MR framework to compute:

 R(A, B) S(B, C)

 Simple extension (proposed independently by multiple

researchers):
 <a, b> from R is mapped as: <b, [R, a]>
 <b, c> from S is mapped as: <b, [S, c]>

 During the reduce phase:

 Join the key-value pairs with the same key but different
relations
EDBT 2011 Tutorial
 How to generalize to:
 R(A, B) S(B, C) T(C,D)
in MapReduce?

EDBT 2011 Tutorial

 U

EDBT 2011 Tutorial

EDBT 2011 Tutorial
 R(A,B) S(B,C) T(C,D)

h(b,c) 0 1 2 3 4 5
0
1
2
3
4
5

EDBT 2011 Tutorial

 Multi-way Join over a single MR phase:
 One-to-many shuffle

 Communication cost:
 3-way Join: O(n) communication
 4-way Join: O(n2)
 …
 M-way Join: O(nM-2)

 Clearly, not feasible for OLAP:

 Large number of Dimension Tables
 Many OLAP queries involve Join of FACT table with multiple
Dimension tables.

EDBT 2011 Tutorial

EDBT 2011 Tutorial
 Observations:
 STAR Schema (or its variant)
 DIMENSION tables typically are not very large (in most cases)
 FACT table, on the other hand, have a large number of rows.

 Design approach:
 Use MapReduce as a distributed & parallel computing substrate
 Fact tables partitioned across multiple nodes
 Partitioning strategy should be based on the dimension that are
most often used as selection constraint in DW queries
 Dimension tables are replicated
 MAP tasks perform the STAR joins
 REDUCE tasks then perform the aggregation

EDBT 2011 Tutorial

 Complex data processing – Graphs and
beyond

 Multidimensional Data Analytics: Location-

based data

 Physical and Virtual Worlds: Social Networks

and Social Media data & analysis

EDBT 2011 Tutorial

 New breed of Analysts:
 Information-savvy users
 Most users will become nimble analysts
 Most transactional decisions will be preceded by a
detailed analysis

 Convergence of OLAP and OLTP:

 Both from the application point-of-view and from
the infrastructure point-of-view
EDBT 2011 Tutorial
 Data in the Cloud
 Platforms for Data Analysis
 Platforms for Update intensive workloads

 Data Platforms for Large Applications

 Multitenant Data Platforms

 Open Research Challenges

EDBT 2011 Tutorial

 Most enterprise solutions are based on RDBMS
technology.
 Significant Operational Challenges:
 Provisioning for Peak Demand
 Resource under-utilization
 Capacity planning: too many variables
 Storage management: a massive challenge
 System upgrades: extremely time-consuming
 Complex mine-field of software and hardware licensing

 Unproductive use of people-resources from a

company’s perspective
EDBT 2011 Tutorial
 Client Site Client Site Client Site

Load Balancer (Proxy)

App App App App App

Server Server Server Server Server

Database Replication the

becomes
MySQL MySQL
Scalability
Master DB BottleneckSlave DB
Cannot leverage elasticity
EDBT 2011 Tutorial
 Client Site Client Site Client Site

Load Balancer (Proxy)

App App App App App

Server Server Server Server Server

Replication
MySQL MySQL
Master DB Slave DB

EDBT 2011 Tutorial

 Client Site Client Site Client Site

Load Balancer (Proxy)

Apache Apache Apache Apache Apache

+ App + App + App + App + App
Server Server Server Server Server

Scalable and Elastic,

Key Value Stores
but limited consistency and
operational flexibility
EDBT 2011 Tutorial
 “If you want vast, on-demand scalability, you need a
non-relational database.” Since scalability
requirements:
 Can change very quickly and,
 Can grow very rapidly.

 Difficult to manage with a single in-house RDBMS

server.
 RDBMS scale well:
 When limited to a single node, but
 Overwhelming complexity to scale on multiple server
nodes.
EDBT 2011 Tutorial
 Initially used for: “Open-Source relational database
that did not expose SQL interface”
 Popularly used for: “non-relational, distributed data
stores that often did not attempt to provide ACID
guarantees”
 Gained widespread popularity through a number of
open source projects
 HBase, Cassandra, Voldemort, MongDB, …

 Scale-out, elasticity, flexible data model, high

availability
EDBT 2011 Tutorial
 Term heavily used (and abused)
 Scalability and performance bottleneck not
inherent to SQL
 Scale-out, auto-partitioning, self-manageability
can be achieved with SQL
 Different implementations of SQL engine for
different application needs
 SQL provides flexibility, portability

EDBT 2011 Tutorial

 Recently renamed
 Encompass a broad category of “structured”
storage solutions
 RDBMS is a subset
 Key Value stores
 Document stores
 Graph database
 The debate on appropriate characterization
continues
EDBT 2011 Tutorial
 Scalability

 Elasticity

 Fault tolerance

 Self Manageability

 Sacrifice consistency?
EDBT 2011 Tutorial
public void confirm_friend_request(user1, user2)
{
begin_transaction();
update_friend_list(user1, user2, [Link]);
//Palo Alto update_friend_list(user2, user1,
[Link]);
//London
end_transaction();
}

EDBT 2011 Tutorial

public void confirm_friend_request_A(user1, user2){
try{ update_friend_list(user1, user2, [Link]); //palo
alto }
catch(exception e){ report_error(e); return; }
try{ update_friend_list(user2, user1, [Link]); //london
}
catch(exception e) { revert_friend_list(user1, user2);
report_error(e); return; }
}

EDBT 2011 Tutorial

public void confirm_friend_request_B(user1, user2){
try{ update_friend_list(user1, user2, [Link]); //palo
alto }catch(exception
e){ report_error(e); add_to_retry_queue([Link]
t, user1, user2, current_time()); }
try{ update_friend_list(user2, user1, [Link]); //london
}catch(exception e)
{ report_error(e); add_to_retry_queue([Link],
user2, user1, current_time()); } }

EDBT 2011 Tutorial

/* get_friends() method has to reconcile results returned by get_friends() because there may be
data inconsistency due to a conflict because a change that was applied from the message
queue is contradictory to a subsequent change by the user. In this case, status is a bitflag
where all conflicts are merged and it is up to app developer to figure out what to do. */
public list get_friends(user1){ list actual_friends = new list(); list friends =
get_friends(); foreach (friend in friends){ if([Link] ==
[Link]){ //no conflict actual_friends.add(friend); }else
if(([Link] &= [Link]) and !([Link] &=
[Link])){ // assume friend is confirmed as long as it wasn’t also
deleted [Link] =
[Link]; actual_friends.add(friend); update_friends
_list(user1, friend, [Link]); }else{ //assume deleted if there is a conflict
with a delete update_friends_list( user1, friend,
[Link]) } }//foreach return actual_friends; }

EDBT 2011 Tutorial

I love eventual consistency but there are some
applications that are much easier to implement with
strong consistency. Many like eventual consistency
because it allows us to scale-out nearly without bound
but it does come with a cost in programming model
complexity.

February 24, 2010

EDBT 2011 Tutorial
 Quest for Scalable, Fault-tolerant,
and Consistent Data Management in
the Cloud that provides
Elasticity

EDBT 2011 Tutorial

 Data in the Cloud

 Data Platforms for Large Applications

 Key value Stores
 Transactional support in the cloud

 Multitenant Data Platforms

 Concluding Remarks

EDBT 2011 Tutorial

 Key-Valued data model
 Key is the unique identifier
 Key is the granularity for consistent access
 Value can be structured or unstructured
 Gained widespread popularity
 In house: Bigtable (Google), PNUTS (Yahoo!),
Dynamo (Amazon)
 Open source: HBase, Hypertable, Cassandra,
Voldemort
 Popular choice for the modern breed of web-
applications

EDBT 2011 Tutorial

 Scale out: designed for scale
 Commodity hardware
 Low latency updates
 Sustain high update/insert throughput
 Elasticity – scale up and down with load
 High availability – downtime implies lost
revenue
 Replication (with multi-mastering)
 Geographic replication
 Automated failure recovery

EDBT 2011 Tutorial

 No Complex querying functionality
 No support for SQL
 CRUD operations through database specific API
 No support for joins
 Materialize simple join results in the relevant row
 Give up normalization of data?
 No support for transactions
 Most data stores support single row transactions
 Tunable consistency and availability (e.g., Dynamo)
 Achieve high scalability

EDBT 2011 Tutorial

 Consistency, Availability, and Network
Partitions
 Only have two of the three together
 Large scale operations – be prepared for
network partitions
 Role of CAP – During a network partition,
choose between Consistency and Availability
 RDBMS choose consistency
 Key Value stores choose availability [low replica
consistency]
EDBT 2011 Tutorial
 It is a simple solution
 nobody understands what sacrificing P means
 sacrificing A is unacceptable in the Web
 possible to push the problem to app developer
 C not needed in many applications
 Banks do not implement ACID (classic example wrong)
 Airline reservation only transacts reads (Huh?)
 MySQL et al. ship by default in lower isolation level
 Data is noisy and inconsistent anyway
 making it, say, 1% worse does not matter
EDBT 2011 Tutorial
[Vogels, VLDB 2007]
 Dynamo – quorum based replication
 Multi-mastering keys – Eventual Consistency
 Tunable read and write quorums
 Larger quorums – higher consistency, lower
availability
 Vector clocks to allow application supported
reconciliation
 PNUTS – log based replication
 Similar to log replay – reliable log multicast
 Per record mastering – timeline consistency
 Major outage might result in losing the tail of the log
EDBT 2011 Tutorial
 A standard benchmarking tool for evaluating
Key Value stores

 Evaluate different systems on common

workloads

 Focus on performance and scale out

EDBT 2011 Tutorial

 Tier 1 – Performance
 Latency versus throughput as throughput increases

 Tier 2 – Scalability
 Latency as database, system size increases
 “Scale-out”

 Latency as we elastically add servers

 “Elastic speedup”

EDBT 2011 Tutorial

 50/50 Read/update

Workload A - Read latency

70

60
Average read latency (ms)

50

40

30

20

10

0
0 2000 4000 6000 8000 10000 12000 14000
Throughput (ops/sec)

Cassandra Hbase PNUTS MySQL

 95/5 Read/update

Workload B - Read latency

20
18
Average read latency (ms)

16
14
12
10
8
6
4
2
0
0 1000 2000 3000 4000 5000 6000 7000 8000 9000
Throughput (operations/sec)

Cassandra HBase PNUTS MySQL

EDBT 2011 Tutorial

 Scans of 1-100 records of size 1KB

Workload E - Scan latency

120

100
Average scan latency (ms)

80

60

40

20

0
0 200 400 600 800 1000 1200 1400 1600
Throughput (operations/sec)

Hbase PNUTS Cassandra

EDBT 2011 Tutorial

 Different databases suitable for different
workloads
 Evolving systems – landscape changing
dramatically
 Active development community around open
source systems

EDBT 2011 Tutorial

 [Dean et al., ODSI 2004] MapReduce: Simplified Data Processing on Large
Clusters, J. Dean, S. Ghemawat, In OSDI 2004
 [Dean et al., CACM 2008] MapReduce: Simplified Data Processing on Large
Clusters, J. Dean, S. Ghemawat, In CACM Jan 2008
 [Dean et al., CACM 2010] MapReduce: a flexible data processing tool, J. Dean, S.
Ghemawat, In CACM Jan 2010
 [Stonebraker et al., CACM 2010] MapReduce and parallel DBMSs: friends or
foes?, M. Stonebraker, D. J. Abadi, D. J. DeWitt, S. Madden, E. Paulson, A. Pavlo,
A, Rasin, In CACM Jan 2010
 [Pavlo et al., SIGMOD 2009] A comparison of approaches to large-scale data
analysis, A. Pavlo et al., In SIGMOD 2009
 [Abouzeid et al., VLDB 2009] HadoopDB: An Architectural Hybrid of MapReduce
and DBMS Technologies for Analytical Workloads, A. Abouzeid et al., In VLDB
2009
 [Afrati et al., EDBT 2010] Optimizing joins in a map-reduce environment, F. N.
Afrati, J. D. Ullman, In EDBT 2010
 [Agrawal et al., SIGMOD 2009] Asynchronous view maintenance for VLSD
databases, P. Agrawal et al., In SIGMOD 2009
 [Das et al., SIGMOD 2010] Ricardo: Integrating R and Hadoop, S. Das et al., In
SIGMOD 2010
 [Cohen et al., VLDB 2009] MAD Skills: New Analysis Practices for Big Data, J.
Cohen et al., In VLDB 2009
EDBT 2011 Tutorial