web-scale data processing
Christopher Olston and many others Yahoo! Research
Motivation
Projects increasingly revolve around analysis of big data sets
Extracting structured data, e.g. face detection Understanding complex large-scale phenomena
social systems (e.g. user-generated web content) economic systems (e.g. advertising markets) computer systems (e.g. web search engines)
Data analysis is inner loop at Yahoo! et al. Big data necessitates parallel processing
Examples
1. Detect faces
You have a function detectFaces() You want to run it over n images n is big
2. Study web usage
You have a web crawl and click log Find sessions that end with the best page
Existing Work
Parallel architectures
cluster computing multi-core processors
Data-parallel software
parallel DBMS Map-Reduce, Dryad
Data-parallel languages
SQL NESL
Pig Project
Data-parallel language (Pig Latin)
Relational data manipulation primitives Imperative programming style Plug in code to customize processing
Various crazy ideas
Multi-program optimization Adaptive data placement Automatic example data generator
Pig Latin Language
[SIGMOD08]
Example 1
Detect faces in many images.
I = load /mydata/images using ImageParser() as (id, image);
F = foreach I generate id, detectFaces(image);
store F into /mydata/faces;
Example 2
Find sessions that end with the best page.
Visits
user Amy Amy Amy Amy Fred url [Link] [Link] [Link] [Link] time 8:00 8:05 10:00 10:05
Pages
url [Link] [Link] [Link] [Link] pagerank 0.9 0.9 0.7 0.2
[Link]/[Link] 12:00
...
...
Efficient Evaluation Method
the answer
...
repartition by user
group-wise processing (identify sessions; examine pageranks) join
...
repartition by url
...
Visits
...
Pages
In Pig Latin
Visits = load /data/visits as (user, url, time);
Visits = foreach Visits generate user, Canonicalize(url), time;
Pages = load /data/pages as (url, pagerank);
VP UserVisits Sessions HappyEndings = = = = join Visits by url, Pages by url;
group VP by user;
foreach UserVisits generate flatten(FindSessions(*));
filter Sessions by BestIsLast(*);
store HappyEndings into '/data/happy_endings';
Pig Latin, in general
transformations on sets of records easy for users
high-level, extensible data processing primitives
easy for the system
exposes opportunities for parallelism and reuse
operators: FILTER FOREACH GENERATE GROUP
binary operators: JOIN COGROUP UNION
Related Languages
SQL: declarative all-in-one blocks NESL: lacks join, cogroup Map-Reduce: special case of Pig Latin
a = FOREACH input GENERATE flatten(Map(*)); b = GROUP a BY $0; c = FOREACH b GENERATE Reduce(*);
Sawzall: rigid map-then-reduce structure
Pig Latin = Sweet Spot Between SQL & Map-Reduce
SQL Programming style Large blocks of declarative Map-Reduce Plug together pipes (but
"I much prefer writing in Pig [Latin] versus SQL. The step-by-step method of
constraints restricted to linear, 2-step creating a program in Pig [Latin] is much cleaner and simpler to use than the single block method of SQL. It is easier to keep track ofdata flow) variables what your are in the process of analyzing Built-in are, and where you Group-by, Sort, Join, Filter, your data.
Sort data Group-by,
manipulations
Execution model
-- Jasmine Novak, Engineer, Yahoo!
Aggregate, Top-k, etc...
Fancy; trust the query Simple, transparent "PIG seems to give optimizer the necessary parallel programming construct
of Map-Reduce] doesnt).
(FOREACH, Opportunities for FLATTEN, COGROUP .. etc) and also give sufcient control in map() Many Few (logic buried back to the programmer (which purely declarative approach reduce()) top automatic optimization and like [SQL on
-- Ricky Ho, Adobe Software
Map-Reduce as Backend
( SQL )
automatic rewrite + optimize or or
user
Pig
Map-Reduce
cluster
details in [VLDB09]
Pig Latin vs. Map-Reduce: Code Reduction
import import import import import import import import import import import import import import import import import import import import import import import import import
Users Fltrd
public class MRExample { public static class LoadPages extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
store
while ([Link]()) { Text t = [Link](); String value = [Link](); if ([Link](0) == '1') [Link]([Link](1)); else [Link]([Link](1));
= load users as (name, age); = filter Users by age >= 18 and age <= 25; Views = load views as (user, url); Jnd = join Fltrd by name, Views by user; Grpd = group Jnd by url; Smmd = foreach Grpd generate group, COUNT(Jnd) as clicks; Srtd = order Smmd by clicks desc; Top5 = limit Srtd 5; store Top5 into top5sites;
[Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; [Link]; } } public static class LoadJoined extends MapReduceBase implements Mapper<Text, Text, Text, LongWritable> { public void map( Text k, Text val, OutputCollector<Text, LongWritable> oc, Reporter reporter) throws IOException { // Find the url String line = [Link](); int firstComma = [Link](','); int secondComma = [Link](',', firstComma); String key = [Link](firstComma, secondComma); // drop the rest of the record, I don't need it anymore, // just pass a 1 for the combiner/reducer to sum instead. Text outKey = new Text(key); [Link](outKey, new LongWritable(1L)); } JobConf lfu = new JobConf([Link]); [Link]("Load and Filter Users"); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link](lfu, new Path("/user/gates/users")); [Link](lfu, new Path("/user/gates/tmp/filtered_users")); [Link](0); Job loadUsers = new Job(lfu); public void map(LongWritable k, Text val, OutputCollector<Text, Text> oc, Reporter reporter) throws IOException { // Pull the key out String line = [Link](); int firstComma = [Link](','); String key = [Link](0, firstComma); String value = [Link](firstComma + 1); Text outKey = new Text(key); // Prepend an index to the value so we know which file // it came from. Text outVal = new Text("1" + value); [Link](outKey, outVal); } } public static class ReduceUrls extends MapReduceBase implements Reducer<Text, LongWritable, WritableComparable, Writable> { public void reduce( Text key, Iterator<LongWritable> iter, OutputCollector<WritableComparable, Writable> oc, Reporter reporter) throws IOException { // Add up all the values we see long sum = 0; while ([Link]()) { sum += [Link]().get(); [Link]("OK"); } } public static class LoadAndFilterUsers extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> { public void map(LongWritable k, Text val, OutputCollector<Text, Text> oc, Reporter reporter) throws IOException { // Pull the key out String line = [Link](); int firstComma = [Link](','); String value = [Link](firstComma + 1); int age = [Link](value); if (age < 18 || age > 25) return; String key = [Link](0, firstComma); Text outKey = new Text(key); // Prepend an index to the value so we know which file // it came from. Text outVal = new Text("2" + value); [Link](outKey, outVal); } [Link](key, new LongWritable(sum)); } } public static class LoadClicks extends MapReduceBase implements Mapper<WritableComparable, Writable, LongWritable, Text> { public void map( WritableComparable key, Writable val, OutputCollector<LongWritable, Text> oc, Reporter reporter) throws IOException { [Link]((LongWritable)val, (Text)key); } } public static class Join extends MapReduceBase implements Reducer<Text, Text, Text, Text> { } public static class LimitClicks extends MapReduceBase implements Reducer<LongWritable, Text, LongWritable, Text> { int count = 0; public void reduce( LongWritable key, Iterator<Text> iter, OutputCollector<LongWritable, Text> oc, Reporter reporter) throws IOException { public void reduce(Text key, Iterator<Text> iter, OutputCollector<Text, Text> oc, Reporter reporter) throws IOException { // For each value, figure out which file it's from and // accordingly. List<String> first = new ArrayList<String>(); List<String> second = new ArrayList<String>(); it // Only output the first 100 records while (count < 100 && [Link]()) { [Link](key, [Link]()); count++; } 18 to } } public static void main(String[] args) throws IOException { JobConf lp = new JobConf([Link]); [Link]("Load Pages"); [Link]([Link]); } }
[Link]; [Link]; [Link]; [Link];
[Link]("OK"); } // Do the cross product and collect the values for (String s1 : first) { for (String s2 : second) { String outval = key + "," + s1 + "," + s2; [Link](null, new Text(outval)); [Link]("OK"); } }
[Link]([Link]); [Link]([Link]); [Link]([Link]); [Link](lp, new Path("/user/gates/pages")); [Link](lp, new Path("/user/gates/tmp/indexed_pages")); [Link](0); Job loadPages = new Job(lp);
JobConf join = new JobConf([Link]); [Link]("Join Users and Pages"); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link](join, new Path("/user/gates/tmp/indexed_pages")); [Link](join, new Path("/user/gates/tmp/filtered_users")); [Link](join, new Path("/user/gates/tmp/joined")); [Link](50); Job joinJob = new Job(join); [Link](loadPages); [Link](loadUsers); JobConf group = new JobConf([Link]); [Link]("Group URLs"); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link](group, new Path("/user/gates/tmp/joined")); [Link](group, new Path("/user/gates/tmp/grouped")); [Link](50); Job groupJob = new Job(group); [Link](joinJob); JobConf top100 = new JobConf([Link]); [Link]("Top 100 sites"); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link]([Link]); [Link](top100, new Path("/user/gates/tmp/grouped")); [Link](top100, new Path("/user/gates/top100sitesforusers18to25")); [Link](1); Job limit = new Job(top100); [Link](groupJob); JobControl jc = new JobControl("Find top 100 sites for users 25"); [Link](loadPages); [Link](loadUsers); [Link](joinJob); [Link](groupJob); [Link](limit); [Link]();
Comparison
180 160 140 120 100 80 60 40 20 0
1/20 the lines of code
Minutes
300 250 200 150 100 50 0
1/16 the development time
Hadoop
Pig
Hadoop
Pig
performance 1.5x Hadoop
Ways to Run Pig
Interactive shell Script file Embed in host language (e.g., Java) soon: Graphical editor
Status
Open-source implementation
[Link] Runs on Hadoop or local machine Active project; many refinements in the works
Wide adoption in Yahoo
100s of users 1000s of Pig jobs/day 60% of ad-hoc Hadoop jobs are via Pig 40% of production jobs via Pig
Status
Gaining traction externally
log processing & aggregation building text indexes collaborative filtering, applied to image & video recommendation systems
"The [Hofmann PLSA E/M] algorithm was implemented in pig in 30-35 lines of pig-latin statements. Took a lot less compared to what it took in implementing the algorithm in Map-Reduce Java. Exactly that's the reason I wanted to try it out in Pig. It took 3-4 days for me to write it, starting from learning pig.
-- Prasenjit Mukherjee, Mahout project
Crazy Ideas
[USENIX08] [VLDB08] [SIGMOD09]
Crazy Idea #1 Multi-Program Optimization
Motivation
User programs repeatedly scan the same data files
web crawl search log
Goal:
Reduce redundant IOs, and hence improve overall system throughput
Approach:
Introduce shared scans capability Careful scheduling of jobs, to maximize benefit of shared scans
Scheduling Shared Scans
7 5 10 4 3 9 1 2 8
Executor
looks at queues and anticipates Scheduler arrivals
(Idle) 6 3
2
not popular
Search Log
Web Crawl
Click Data
Crazy Idea #2 Adaptive Data Placement
Motivation
Hadoop is good at localizing computation to data, for conventional map-reduce scenarios However, advanced query processing operations change this:
Pre-hashed join of A & B: co-locate A & B? Frag-repl join of A & B: more replicas of B?
Our idea:
Adaptive pressure-based mechanism to move data s.t. better locality arises
Adaptive Data Placement
jobs
Job Localizer
Scan C Scan C Scan C execution engine Join A A B Scan &
A C
A D
Worker 1
Worker 2
Crazy Idea #3 Automatic Example Generator
Example Program (abstract view)
Find users who tend to visit good pages.
Load Visits(user, url, time) Load Pages(url, pagerank)
Transform to (user, Canonicalize(url), time) Join url = url
Group by user
Transform to (user, Average(pagerank) as avgPR)
Filter avgPR > 0.5
Load Visits(user, url, time) (Amy, [Link], 8am) (Amy, [Link] 9am) (Fred, [Link]/[Link], 11am) Transform to (user, Canonicalize(url), time) Join url = url (Amy, [Link], 8am) (Amy, [Link], 9am) (Fred, [Link], 11am)
Load Pages(url, pagerank)
([Link], 0.9) ([Link], 0.4)
(Amy, [Link], 8am, 0.9) (Amy, [Link], 9am, 0.4) (Fred, [Link], 11am, 0.4) Group by user (Amy, { (Amy, [Link], 8am, 0.9), (Amy, [Link], 9am, 0.4) }) (Fred, { (Fred, [Link], 11am, 0.4) })
Transform to (user, Average(pagerank) as avgPR) (Amy, 0.65) (Fred, 0.4) Filter avgPR > 0.5 (Amy, 0.65)
Automatic Example Data Generator
Objectives:
Realism Conciseness Completeness
Challenges:
Large original data Selective operators (e.g., join, filter) Noninvertible operators (e.g., UDFs)
Talk Summary
Data-parallel language (Pig Latin)
Sequence of data transformation steps Users can plug in custom code
Research nuggets
Joint scheduling of related programs, to amortize IOs Adaptive data placement, to enhance locality Automatic example data generator, to make users life easier
Credits
Yahoo! Grid Team project leads: Alan Gates Olga Natkovich
Yahoo! Research project leads: Chris Olston Utkarsh Srivastava Ben Reed