VNUHCM - University of Science
Faculty of Information Technology
CS435 – Lecture 2
MapReduce
Le Thi Nhan
ltnhan@[Link]
Designed by SlidesCarnival
Content
» Introduction
» Distributed File System
» Computation model
» Example
» Details of execution
» Refinements
» Ref. CS246 – Mining Massive Datasets, Stanford
University
Intro2 Big Data Analytics - FIT - HCMUS 2
1.
Introduction
What will we do with Big Data?
Review
» You know what Big Data is
» You also understand why we need new
technologies/methods
⋄ Acquisition
⋄ Organization
⋄ Analysis
⋄ Decision
Intro2 Big Data Analytics - FIT - HCMUS 4
Do we need super-computers?
Intro2 Big Data Analytics - FIT - HCMUS 5
How should we compute?
» In a large scale computing, one may think
⋄ Distributed vs. Parallel
From wikipedia
Intro2 Big Data Analytics - FIT - HCMUS 6
Google example (2014)
» 10+ billion web pages
» Average size of a webpage = 20KB
» 10 billion * 20KB = 200+ TB
» Disk read bandwidth = 50MB/sec
» Time to read = 4 million seconds > 46 days
» May take longer to do something useful with
the data
Intro2 Big Data Analytics - FIT - HCMUS 7
Cluster computing
Switch
2-10 Gbps backbone
From [Link]
between racks
Switch 1 Gbps between Switch
any pair of nodes
in a rack
CPU CPU CPU CPU
Mem … Mem Mem … Mem
Disk Disk Disk Disk
Intro2 Big Data Analytics - FIT - HCMUS Each rack contains 16-64 nodes Single node 8
Google had 900,000 machines in 2011
Intro2 Big Data Analytics - FIT - HCMUS 9
Large-scale computing
» Machines fail
⋄ 1 server may stay up 3 years
⋄ 1,000 servers in cluster
− Expect to lose 1 server/day
⋄ ~1M servers in cluster
− 1,000 servers fail every day
» Network bottleneck
⋄ Network bandwidth = 1 Gbps
⋄ Moving 10TB takes approximately 1 day
Intro2 Big Data Analytics - FIT - HCMUS 10
Large-scale computing (cont.)
» Challenges
⋄ How to store data persistently and keep it
available if nodes fail?
⋄ How to deal with node failures during a long-
running computation?
⋄ Files must be stored redundantly
⋄ How do we distribute computation?
⋄ How do we make it easy to write distributed
programs?
⋄ Computations must be divided into tasks
Intro2 Big Data Analytics - FIT - HCMUS 11
Large-scale computing (cont.)
» Elegant way to work with big data
⋄ Storage infrastructure – File system
− Google file system (GFS)
− Hadoop distributed file system (HDFS): an open-
source DFS is distributed by Apache
− CloudStore: an open-source DFS is developed by
Kosmix
⋄ Programming model
− MapReduce
Intro2 Big Data Analytics - FIT - HCMUS 12
Hadoop/MapReduce
» Addresses these challenges
⋄ Store data redundantly
− On multiple nodes for persistence and availability
⋄ Move computation close to data
− Minimizing data movement
⋄ Simple programming model
− Hide the complexity
Intro2 Big Data Analytics - FIT - HCMUS 13
2.
Distributed File System
DFS
Introduction
» To exploit cluster computing
⋄ Files must behave somewhat differently from the
conventional file systems
» Distributed file system
⋄ Provides global file namespace, redundancy and
availability
» Typical usage pattern
⋄ Huge files (hundreds of GB to TB)
⋄ Data is rarely updated in place
⋄ Reads and appends are common
Intro2 Big Data Analytics - FIT - HCMUS 15
DFS
» Files are divided into “chunks”
» Chunks are replicated at different nodes
C0 C1 D0 C1 C2 C5 C0 C5
C5 C2 C5 C3 D0 D1 … D0 C2
Chunk server 1 Chunk server 2 Chunk server 3 Chunk server N
Bring computation directly to the data!
Chunk servers also serve as compute servers
Intro2 Big Data Analytics - FIT - HCMUS 16
DFS (cont.)
» Chunk server
⋄ File is split into contiguous chunks (16-64MB)
⋄ Each chunk replicated (usually 2x or 3x)
⋄ Try to keep replicas in different racks
» Master node
⋄ Name Node in Hadoop’s HDFS
⋄ Stores metadata about where files are stored
⋄ Might be replicated
» Client library for file access
⋄ Talks to master to find chunk servers
⋄ Connects directly to chunk servers to access data
Intro2 Big Data Analytics - FIT - HCMUS 17
3.
Computation model
MapReduce
Example
» We have a huge text document
⋄ File is too large for memory
» Count the number of times each distinct word
appears in the file
» Application
⋄ Analyze web server logs to find popular URLs
Intro2 Big Data Analytics - FIT - HCMUS 19
Overview
» (1) Map
⋄ Scan input file record
⋄ Extract something you care about from each
record (keys)
» (2) Group by key
⋄ Sort and shuffle
» (3) Reduce
⋄ Aggregate, summarize, filter or transform
⋄ Write the result
Intro2 Big Data Analytics - FIT - HCMUS 20
Overview (cont.)
From [Link]
Intro2 Big Data Analytics - FIT - HCMUS 21
(1) Map task
» A chunk
⋄ A collection of elements
− Eg. tuples, documents…
− No element is stored across two chunks
» Map function
⋄ Input : an element
⋄ Output : key-value pairs
Intro2 Big Data Analytics - FIT - HCMUS 22
(1) Map task – Example
» Input file
⋄ A collection of documents
⋄ Each document is an element
» Map function
⋄ (key:word, value:integer)
⋄ Read a document and break it into a sequence
of words w1, w2, …, wn
⋄ Sequence of key-value pairs
− (w1,1), (w2,1), …, (wn,1)
Intro2 Big Data Analytics - FIT - HCMUS 23
(1) Map task – Note
» A single map-task will process many
documents
⋄ Its output will be more than one sequence
» If a word w appears m times among all the
documents assigned to that process
⋄ There will be m key-value pairs (w,1) among its
output
Intro2 Big Data Analytics - FIT - HCMUS 24
(2) Grouping by key
» As soon as map-tasks have all completed
⋄ Key-value pairs are grouped by key
⋄ Values associated with each key are formed
into a list of values
» Grouping is performed by the system
⋄ For each key k
⋄ A pair of the form (k, [v1, v2, …, vn]) is produced
Intro2 Big Data Analytics - FIT - HCMUS 25
(3) Reduce task
» Reduce function
⋄ Input : a pair of key and its list of associated
values
⋄ Output : a sequence of key-value pairs
Intro2 Big Data Analytics - FIT - HCMUS 26
(3) Reduce task – Example
» Reduce function
⋄ Adding up all values of a key
⋄ Output: a sequence of (w, m)
− w : a word that appear at least once among all the
input documents
− m : the total number of occurrences of w among all
those documents
Intro2 Big Data Analytics - FIT - HCMUS 27
(3) Reduce task – Note
» When reduce function is associative and
commutative
⋄ The values can be combined in any order, with
the same result
» We can push reduce function within the map task
⋄ Instead of the map tasks producing pairs (w,1),
(w,1)…
⋄ Map-tasks would be replaced by a pair (w,m)
» It’s still necessary to do grouping and pass the
result to reduce tasks
Intro2 Big Data Analytics - FIT - HCMUS 28
In brief
Input Intermediate
key-value pairs key-value pairs
k v
map
v
k
k v
map
v
k k v
… …
map
v k v
k
Intro2 Big Data Analytics - FIT - HCMUS From [Link] 29
In brief (cont.)
Intermediate Output
key-value pairs Key-value groups key-value pairs
reduce
k v v v k v
k v
group reduce
k v by key k v v k v
k v
… … …
k v k v
k v
Intro2 Big Data Analytics - FIT - HCMUS From [Link] 30
In brief (cont.)
MAP: GROUP by key: REDUCE:
Read input and Collect all pairs Collect all values
produces a set of with same key belonging to the
key-value pairs key and output
The crew of the space
shuttle Endeavor
recently returned to
Earth as ambassadors,
(The, 1) (crew, 1)
harbingers of a new era (crew, 1) (crew, 1)
Only sequential reads
of space exploration. (crew, 2)
(of, 1) (space, 1)
Scientists at NASA are (space, 1)
saying that the recent (the, 1) (the, 1)
assembly of the Dextre (the, 3)
bot is the first step in a (space, 1) (the, 1)
(shuttle, 1)
long-term space-based (shuttle, 1) (the, 1)
man/mache (recently, 1)
partnership. '"The work (Endeavor, 1) (shuttle, 1)
we're doing now -- the …
robotics we're doing --
(recently, 1) (recently, 1)
is what we're going to …. …
need
……………………..
Big document (key, value) (key, value) (key, value)
Intro2 Big Data Analytics - FIT - HCMUS From [Link] 31
In brief (cont.)
map(key, value)
// key: document name;
// value: text of the document
for each word w in value:
emit(w, 1)
reduce(key, values)
// key: a word;
// value: an iterator over counts
result = 0
for each count v in values:
result += v
emit(key, result)
Intro2 Big Data Analytics - FIT - HCMUS 32
4.
Examples
1 – Host size
» Suppose we have a large web corpus with a
metadata file formatted as follows
⋄ Each record of the form (URL, size, date, …)
» For each host, find the total number of bytes
⋄ The sum of the page sizes for all URLs from a particular
host
» Map
⋄ For each record, output (hostname(URL), size)
» Reduce
⋄ Sum of size for each host
Intro2 Big Data Analytics - FIT - HCMUS 34
2 – Language model
» Count number of times each 5-word
sequence occurs in a large corpus of
documents
» Map
⋄ Extract (5-word sequence, count) from
document
» Reduce
⋄ Combine the counts
Intro2 Big Data Analytics - FIT - HCMUS 35
3 – Join operation
» Compute the natural join R(A,B) ⋈ S(B,C)
⋄ R and S are each stored in files
⋄ Tuples are pairs (a,b) or (b,c)
A B B C A C
a1 b1 b2 c1 a3 c1
a2 b1 ⋈ b2 c2 = a3 c2
a3 b2 b3 c3 a4 c3
a4 b3
R S
Intro2 Big Data Analytics - FIT - HCMUS 36
3 – Join operation (cont.)
» Use a hash function h from B-values to 1…k
» Map process
⋄ Turn each input tuple
− R(a,b) into key-value pair (b,(a,R))
− S(b,c) into key-value pair (b,(c,S))
⋄ Send each key-value pair with key b to Reduce
process h(b)
− Hadoop does this automatically, just tell it what k is
» Reduce process
⋄ Match all the pair (b,(a,R)) with all (b,(c,S))
⋄ And outputs (a,b,c)
Intro2 Big Data Analytics - FIT - HCMUS 37
In brief
» MapReduce is great for
⋄ Problems that require sequential data access
⋄ Large batch jobs (not interactive, real-time)
» MapReduce is inefficient for
⋄ Problems where random access to data
required
− Graphs
− Interdependent data (comparisons of many pairs of
items)
Intro2 Big Data Analytics - FIT - HCMUS 38
5.
Details of execution
Overview
From [Link]
Intro2 Big Data Analytics - FIT - HCMUS 40
MapReduce – Environment
» Takes care of
⋄ Partitioning the input data
⋄ Scheduling the program’s execution across a
set of machines
⋄ Performing the group by key step
⋄ Handling machine failures
⋄ Managing required inter-machine
communication
Intro2 Big Data Analytics - FIT - HCMUS 41
MapReduce – Diagram
Big document
MAP:
Read input and
produces a set of
key-value pairs
GROUP by key:
Collect all pairs with
same key
(Hash merge,
Shuffle, Sort,
Partition)
REDUCE:
Collect all values
belonging to the key
and output
Intro2 Big Data Analytics - FIT - HCMUS From [Link] 42
MapReduce – In parallel
From [Link]
All phases are distributed with many tasks doing the work
Intro2 Big Data Analytics - FIT - HCMUS 43
MapReduce – Workflow
Input 0 Input 1 Input 2
» Read inputs as a set of key-
value pairs
» Map: transform input 𝑘𝑣-pairs
into a new set of 𝑘’𝑣’-pairs Map 0 Map 1 Map 2
» Group by key: sort and
shuffle the 𝑘’𝑣’-pairs to output
nodes Shuffle
» All 𝑘’𝑣’-pairs with a given 𝑘’ are
sent to the same reduce
» Reduce: process all 𝑘’𝑣’-pairs Reduce 0 Reduce 1
grouped by key into new 𝑘”𝑣”-
pairs
» Write the resulting pairs to files
Out 0 Out 1
Intro2 Big Data Analytics - FIT - HCMUS From [Link] 44
MapReduce – Workflow (cont.)
» Input & final output are stored on the DFS
⋄ Scheduler tries to schedule map tasks “close”
to physical storage location of input data
» Intermediate results are stored on local file
system of map & reduce workers
» Output is often input to another MapReduce
task
Intro2 Big Data Analytics - FIT - HCMUS 45
MapReduce – Master node
» Takes care of coordination
⋄ Task status (idle, in-progress, completed)
⋄ Idle tasks
− Get scheduled as workers become available
⋄ When a map task completes
− It sends the master the location and size of
intermediate files, one for each reducer
⋄ Master pushes this into reducers
» Master pings workers periodically to detect
failures
Intro2 Big Data Analytics - FIT - HCMUS 46
Dealing with failure
» Map worker failure
⋄ Map tasks completed or in-progress at worker
are reset to idle
⋄ Idle tasks rescheduled on other workers
» Reduce worker failure
⋄ Only in-progress tasks are reset to idle
⋄ Idle reduce tasks restarted on other workers
» Master failure
⋄ MapReduce task is aborted and client is notified
Intro2 Big Data Analytics - FIT - HCMUS 47
How many MapReduce tasks?
» 𝑀 map tasks, 𝑅 reduce tasks
» Rule of thumb
⋄ Make 𝑀 much larger than the number of nodes in
the cluster
⋄ One DFS chunk per map is common
⋄ Improve dynamic load balancing and speed up
recovery from worker failures
» Usually 𝑅 is smaller than 𝑀
⋄ Because output is spread across 𝑅 files
Intro2 Big Data Analytics - FIT - HCMUS 48
6.
Refinements
Combiners
» A map task will produce many pair of the
form 𝑘, 𝑣1 , (𝑘, 𝑣2)… for the same 𝑘
» Can save network time by pre-aggregating
values in the mapper
» Combine (𝑘, 𝑙𝑖𝑠𝑡(𝑣1)) → 𝑣2
» Combiner is usually same as the reduce
function
Intro2 Big Data Analytics - FIT - HCMUS 51
Example
» Combiner combines the values of all keys of
a single mapper
» Much less data needs to be copied and
shuffled
From [Link]
Intro2 Big Data Analytics - FIT - HCMUS 52
Combiners (cont.)
» Combiners trick works only if reduce function
is commutative and associative
» Sum ?
» Average ?
» Median ?
Intro2 Big Data Analytics - FIT - HCMUS 53
THANKS!
Any questions?
55