“Hadoop”:

A Distributed Architecture, FileSystem, & MapReduce

Big Data Analytics, The Class

Goal: Generalizations
A model or summarization of the data.

Data Frameworks Algorithms and Analyses

Similarity Search
Hadoop File System Spark Hypothesis Testing
Streaming Graph Analysis
MapReduce Recommendation Systems
Tensorflow
DeepLearning

1
 Big Data Analytics, The
Class

Big Data Analytics

Big Data Analytics, The Class

Big Data Analytics

2
 Classical Data Analytics

CPU

Memory

Disk

Classical Data Analytics

CPU

Memory
(64 GB)

Disk

3
 Classical Data Analytics

CPU

Memory
(64 GB)

Disk

Classical Data Analytics

CPU

Memory
(64 GB)

Disk

4
 IO Bounded
Reading a word from disk versus main memory: 105 slower!
Reading many contiguously stored words
is faster per word, but fast modern disks
still only reach 150MB/s for sequential reads.

IO Bounded
Reading a word from disk versus main memory: 105 slower!
Reading many contiguously stored words
is faster per word, but fast modern disks
still only reach 150MB/s for sequential reads.

IO Bound: biggest performance bottleneck is reading / writing to disk.

starts around 100 GBs: ~10 minutes just to read

200 TBs: ~20,000 minutes = 13 days

10

5
 Classical Big Data

CPU
Classical focus: efficient use of disk.
e.g. Apache Lucene / Solr
Memory

Disk
Classical limitation: Still bounded when
needing to process all of a large file.

11

Classical Big Data

Howto solve?
Classical limitation: Still bounded when needing to
process all of a large file.

12

6
 Distributed Architecture
Switch
~10Gbps

Rack 1
Rack 2
Switch Switch
~1Gbps ~1Gbps
...

CPU CPU CPU CPU CPU CPU

Memory Memory ... Memory Memory Memory ... Memory

Disk Disk Disk Disk Disk Disk

13

Distributed Architecture
In reality, modern setups often have multiple cpus and disks
per server, but we will model as if one machine
per cpu-disk pair.
Switch
~1Gbps

CPU CPU CPU CPU CPU CPU

... ...
...
Memory Memory

Disk Disk ... Disk Disk Disk ... Disk

14

7
 Distributed Architecture (Cluster)

Switch
~10Gbps

Rack 1
Rack 2
Switch Switch
~1Gbps ~1Gbps
...

CPU CPU CPU CPU CPU CPU

Memory Memory ... Memory Memory Memory ... Memory

Disk Disk Disk Disk Disk Disk

15

Distributed Architecture (Cluster)

Challenges for IO Cluster Computing

1. Nodes fail
1 in 1000 nodes fail a day

2. Network is a bottleneck
Typically 1-10 Gb/s throughput

3. Traditional distributed programming is

often ad-hoc and complicated

16

8
 Distributed Architecture (Cluster)
Challenges for IO Cluster Computing
1. Nodes fail
1 in 1000 nodes fail a day
Duplicate Data
2. Network is a bottleneck
Typically 1-10 Gb/s throughput
Bring computation to nodes, rather than
data to nodes.
3. Traditional distributed programming is
often ad-hoc and complicated
Stipulate a programming system that
can easily be distributed

17

Distributed Architecture (Cluster)

Challenges for IO Cluster Computing
1. Nodes fail
1 in 1000 nodes fail a day
Duplicate Data
2. Network is a bottleneck
Typically 1-10 Gb/s throughput HDFS with
Bring computation to nodes, rather than
MapReduce
data to nodes.
accomplishes
3. Traditional distributed programming is
often ad-hoc and complicated
all!
Stipulate a programming system that
can easily be distributed

18

9
 Distributed Filesystem

The effectiveness of MapReduce, Spark, and other

distributed processing systems is in part simply due to
use of a distributed filesystem!

19

Distributed Filesystem
Characteristics for Big Data Tasks
Large files (i.e. >100 GB to TBs)
Reads are most common
No need to update in place
(append preferred)
CPU

Memory

Disk

20

10
 Distributed Filesystem

(e.g. Apache HadoopDFS, GoogleFS, EMRFS)

C, D: Two different files

[Link]
-apache-hadoop-big-data

C
D

21

Distributed Filesystem “Hadoop”was namedafter atoy

elephant belonging to Doug
Cutting’s son. Cutting was oneof
(e.g. Apache HadoopDFS, GoogleFS,
EMRFS) Hadoop’s creators.
C, D: Two different files
[Link]
-apache-hadoop-big-data

C
D

22

11
 Distributed Filesystem

(e.g. Apache HadoopDFS, GoogleFS, EMRFS)

C, D: Two different files; break into chunks (or "partitions"):

C0 D0

C1 D1

C2 D2

C3 D3

C4 D4

C5 D5

23

Distributed Filesystem

(e.g. Apache HadoopDFS, GoogleFS, EMRFS)

C, D: Two different files

chunk server 1 chunk server 2 chunk server 3 chunk server n

(Leskovec at al., 2014; [Link]

24

12
 Distributed Filesystem

(e.g. Apache HadoopDFS, GoogleFS, EMRFS)

C, D: Two different files

chunk server 1 chunk server 2 chunk server 3 chunk server n

(Leskovec at al., 2014; [Link]

25

Distributed Filesystem

(e.g. Apache HadoopDFS, GoogleFS, EMRFS)

C, D: Two different files

chunk server 1 chunk server 2 chunk server 3 chunk server n

(Leskovec at al., 2014; [Link]

26

13
 Distributed Filesystem
Chunk servers (on Data Nodes)
File is split into contiguous chunks
Typically each chunk is 16-64MB
Each chunk replicated (usually 2x or 3x)
Try to keep replicas in different racks

(Leskovec at al., 2014; [Link]

27

Components of a Distributed Filesystem

Chunk servers (on Data Nodes)
File is split into contiguous chunks
Typically each chunk is 16-64MB
Each chunk replicated (usually 2x or 3x)
Try to keep replicas in different racks
Name node (aka master node)
Stores metadata about where files are stored
Might be replicated or distributed across data nodes.

(Leskovec at al., 2014; [Link]

28

14
 Components of a Distributed Filesystem
Chunk servers (on Data Nodes)
File is split into contiguous chunks
Typically each chunk is 16-64MB
Each chunk replicated (usually 2x or 3x)
Try to keep replicas in different racks
Name node (aka master node)
Stores metadata about where files are stored
Might be replicated or distributed across data nodes.
Client library for file access
Talks to master to find chunk servers
Connects directly to chunk servers to access data

(Leskovec at al., 2014; [Link]

29

Distributed Architecture (Cluster)

Challenges for IO Cluster Computing
1. Nodes fail
1 in 1000 nodes fail a day
Duplicate Data (Distributed FS)
2. Network is a bottleneck
Typically 1-10 Gb/s throughput
Bring computation to nodes, rather than
data to nodes.
3. Traditional distributed programming is
often ad-hoc and complicated
Stipulate a programming system that
can easily be distributed

30

15
 What is MapReduce

noun.1 - A style of programming

input chunks => map tasks | group_by keys | reduce tasks => output

“|” is the linux “pipe” symbol: passes stdout from first process to stdin of next.

31

What is MapReduce

noun.1 - A style of programming

input chunks => map tasks | group_by keys | reduce tasks => output

“|” is the linux “pipe” symbol: passes stdout from first process to stdin of next.

E.g. counting words:

tokenize(document) | sort | uniq -c

32

16
 What is MapReduce

noun.1 - A style of programming

input chunks => map tasks | group_by keys | reduce tasks => output

“|” is the linux “pipe” symbol: passes stdout from first process to stdin of next.

E.g. counting words:

tokenize(document) | sort | uniq -c

noun.2 - A system that distributes MapReduce style programs across a

distributed file-system.

(e.g. Google’s internal “MapReduce” or [Link] with hdfs)

33

What is MapReduce

34

17
 What is MapReduce

extract what
you care
about.

line => (k, v) Map

35

What is MapReduce

sort and
shuffle

many (k, v) =>

(k, [v1, v2]), ...
extract what
you care
about.

Map

36

18
 What is MapReduce

sort and
shuffle
extract what
you care
about. aggregate,
summarize
Map
Reduce

37

What is MapReduce

Easy as 1, 2, 3!
Step 1: Map Step 2: Sort / Groupby Step 3: Reduce

38

19
 What is MapReduce

Easy as 1, 2, 3!
Step 1: Map Step 2: Sort / Groupby Step 3: Reduce

(Leskovec at al., 2014; [Link]

39

(1) The Map Step

(Leskovec at al., 2014; [Link]

40

20
 (2) The Sort / Group-by Step

(Leskovec at al., 2014; [Link]

41

(3) The Reduce Step

(Leskovec at al., 2014; [Link]

42

21
 What is MapReduce

Easy as 1, 2, 3!
Step 1: Map Step 2: Sort / Groupby Step 3: Reduce

(Leskovec at al., 2014; [Link]

43

What is MapReduce

Map: (k,v) -> (k’, v’)*

(Written by programmer)

Group by key: (k1’, v1’), (k2’, v2’), ... -> (k1’, (v1’, v’, …),
(system handles) (k2’, (v1’, v’, …), …

Reduce: (k’, (v1’, v’, …)) -> (k’, v’’)*

(Written by programmer)

44

22
 Example: Word Count
tokenize(document) | sort | uniq -c

45

Example: Word Count

tokenize(document) | sort | uniq -c

Map: extract
what you sort and Reduce:
care about. shuffle aggregate,
summarize

46

23
 Example: Word Count

(Leskovec at al., 2014; [Link]

47

48

24
49

50

25
 (Leskovec at al., 2014;
[Link]

Chunks

51

Example: Word Count

@abstractmethod
def map(k, v):
pass

@abstractmethod
def reduce(k, vs):
pass

52

26
 Example: Word Count (v1)

def map(k, v):

for w in tokenize(v):
yield (w,1)

def reduce(k, vs):

return len(vs)

53

Example: Word Count (v1)

def map(k, v): def tokenize(s):
for w in tokenize(v): #simple version
yield (w,1) return [Link](‘ ‘)

def reduce(k, vs):

return len(vs)

54

27
 Example: Word Count (v2)
def map(k, v):
counts = dict()
for w in tokenize(v):
counts each word within the chunk
(try/except is faster than
“if w in counts”)

55

Example: Word Count (v2)

def map(k, v):

counts = dict()
for w in tokenize(v):
try:
counts[w] += 1 counts each word within the chunk
except KeyError: (try/except is faster than
counts[w] = 1 “if w in counts”)
for item in [Link]():
yield item

56

28
 Example: Word Count (v2)

def map(k, v):

counts = dict()
for w in tokenize(v):
try:
counts[w] += 1 counts each word within the chunk
except KeyError: (try/except is faster than
counts[w] = 1 “if w in counts”)
for item in [Link]():
yield item

def reduce(k, vs): sum of counts from different chunks

return (k, sum(vs) )

57

Distributed Architecture (Cluster)

Challenges for IO Cluster Computing
1. Nodes fail
1 in 1000 nodes fail a day
Duplicate Data (Distributed FS)
2. Network is a bottleneck
Typically 1-10 Gb/s throughput
Bring computation to nodes, rather than
data to nodes.
3. Traditional distributed programming is
often ad-hoc and complicated
Stipulate a programming system that
can easily be distributed

58

29
 Distributed Architecture (Cluster)
Challenges for IO Cluster Computing
1. Nodes fail
1 in 1000 nodes fail a day
Duplicate Data (Distributed FS)
2. Network is a bottleneck
Typically 1-10 Gb/s throughput
Bring computation to nodes, rather than
data to nodes. (Sort and Shuffle)
3. Traditional distributed programming is
often ad-hoc and complicated
Stipulate a programming system that
can easily be distributed

59

Distributed Architecture (Cluster)

Challenges for IO Cluster Computing
1. Nodes fail
1 in 1000 nodes fail a day
Duplicate Data (Distributed FS)
2. Network is a bottleneck
Typically 1-10 Gb/s throughput
Bring computation to nodes, rather than
data to nodes. (Sort and Shuffle)
3. Traditional distributed programming is
often ad-hoc and complicated (Simply define a map
Stipulate a programming system that and reduce)
can easily be distributed

60

30
 Example: Relational Algebra

Select

Project
Union, Intersection, Difference

Natural Join

Grouping

61

Example: Relational Algebra

Select

Project

Union, Intersection, Difference

Natural Join

Grouping

62

31
 Example: Relational Algebra

Select

R(A1,A2,A3,...), Relation R, Attributes A*

return only those attribute tuples where condition C is true

63

Example: Relational Algebra

Select
R(A1,A2,A3,...), Relation R, Attributes A*
return only those attribute tuples where condition C is true
def map(k, v): #v is list of attribute tuples: [(...,), (...,), ...]
r = []
for t in v:
if t satisfies C:
r += [(t, t)]
return r

64

32
 Example: Relational Algebra

Select
R(A1,A2,A3,...), Relation R, Attributes A*
return only those attribute tuples where condition C is true
def map(k, v): #v is list of attribute tuples: [(...,), (...,), ...]
r = []
for t in v:
if t satisfies C:
r += [(t, t)]
return r
def reduce(k, vs):
r = []
for each v in vs:
r += [(k, v)]
return r

65

Example: Relational Algebra

Select

R(A1,A2,A3,...), Relation R, Attributes A*

return only those attribute tuples where condition C is true

def map(k, v): #v is list of attribute tuples
for t in v:
if t satisfies C:
yield (t, t)

def reduce(k, vs):

For each v in vs:
yield (k, v)

66

33
 Example: Relational Algebra

Natural Join
Given R1 and R2 return Rjoin
-- union of all pairs of tuples that match given attributes.
def map(k, v): #k \in {R1, R2}, v is (A, B) for R1, (B, C) for
R2 #B are matched attributes

67

Example: Relational Algebra

Natural Join
Given R1 and R2 return Rjoin
-- union of all pairs of tuples that match given attributes.
def map(k, v): #k \in {R1, R2}, v is (A, B) for R1, (B, C) for
R2 #B are matched attributes
if k==’R1’:
(a, b) = v
return (b,(‘R1’,a))
if k==’R2’:
(b,c) = v
return (b,(‘R2’,c))

68

34
 Example: Relational Algebra

Natural Join
Given R1 and R2 return Rjoin
-- union of all pairs of tuples that match given attributes.
def map(k, v): #k \in {R1, R2}, v is (A, B) for R1, (B, C) for R2
#B are matched attributes
if k==’R1’:
def reduce(k, vs):
(a, b) = v
return (b,(‘R 1’,a)) r1, r2, rjn = [], [], []
if k==’R2’: for (s, x) in vs: #separate rs
(b,c) = v if s == ‘R1’: [Link](x)
return (b,(‘R2’,c)) else: [Link](x)
for a in r1: #join as tuple
for each c in r2:
rjn += (‘Rjoin’, (a, k, c)) #k is b
return rjn

69

Data Flow

70

35
Data Flow

hash

(Leskovec at al., 2014; [Link]

71

Data Flow

Programmer

hash

Programmer

(Leskovec at al., 2014; [Link]

72

36
 Data Flow

DFS Map Map’s Local FS Reduce DFS

73

Data Flow

MapReduce system handles:

● Partitioning
● Scheduling map / reducer execution
● Group by key

● Restarts from node failures

● Inter-machine communication

74

37
 Data Flow

DFS MapReduce DFS

● Schedule map tasks near physical storage

of chunk
● Intermediate results stored locally
● Master / Name Node coordinates

75

Data Flow

DFS MapReduce DFS

● Schedule map tasks near physical storage

of chunk
● Intermediate results stored locally
● Master
○ Task/status:
Name Node
idle, coordinates
in-progress, complete
○ Receives location of intermediate results and schedules with reducer
○ Checks nodes for failures and restarts when necessary
■ All map tasks on nodes must be completely restarted
■ Reduce tasks can pickup with reduce task failed

76

38
 Data Flow

DFS MapReduce DFS

● Schedule map tasks near physical storage

of chunk
● Intermediate results stored locally
● Master / Name Node coordinates
○ Task status: idle, in-progress, complete
○ Receives location of intermediate results and schedules with reducer
○ Checks nodes for failures and restarts when necessary

■ All map tasks on nodes must be completely restarted

■ Reduce tasks can pickup with reduce task failed

DFS MapReduce DFS MapReduce DFS

77

Data Flow

Skew: The degree to which certain tasks end up taking much

longer than others.

Handled with:

● More reducers than reduce tasks

● More reduce tasks than nodes

78

39
 Data Flow

Key Question: Howmany Map and Reduce jobs?

M: map tasks, R: reducer tasks

79

Data Flow

Key Question: Howmany Map and Reduce jobs?

M: map tasks, R: reducer tasks

Answer: 1) If possible, one chunk per map task, and

2) M >> |nodes| ≈≈ |cores|
(better handling of node failures, better load balancing)
3) R <= M
(reduces number of parts stored in DFS)

80

40
Data Flow Reduce Task
version 1: few reduce tasks
(same number of reduce tasks as nodes)

node1

node2

node3

node4

node5

time
Reduce tasks represented by
time to complete task
(some tasks take much longer)

81

Data Flow Reduce Task

version 1: few reduce tasks version 2: more reduce tasks
(same number of reduce tasks as nodes) (more reduce tasks than nodes)

node1 node1

node2 node2

node3 node3

node4 node4

node5 node5

time time
Reduce tasks represented by Reduce tasks represented by
time to complete task time to complete task
(some tasks take much longer) (some tasks take much longer)

82

41
Data Flow Reduce Task
version 1: few reduce tasks version 2: more reduce tasks
(same number of reduce tasks as nodes) (more reduce tasks than nodes)

node1 node1 node1

Last task
completed
node2 node2 node2

node3 node3 node3

Can
redistribute
these tasks to
node4 node4 other nodes node4

node5 node5 node5

time time
Reduce tasks represented by Reduce tasks represented by time
time to complete task time to complete task (the last task now completes
(some tasks take much longer) (some tasks take much longer) much earlier )

83

Communication Cost Model

How to assess performance?

(1) Computation: Map + Reduce + System Tasks

(2) Communication: Moving (key, value) pairs

84

42
 Communication Cost Model

How to assess performance?

(1) Computation: Map + Reduce + System Tasks

(2) Communication: Moving (key, value) pairs

Ultimate Goal: wall-clock Time.

85

Communication Cost Model

How to assess performance?

(1) Computation: Map + Reduce + System Tasks

● Mappers and reducers often single pass O(n) within node
(2) C●om
Sym umn:icsoarttitohne :keMysoivsiunsguaklle
ste yym,ov
staelu
xpeenpsa
iviers
● Even if map executes on same node, disk read usually
dominates
● In any case, can add more nodes
Ultimate Goal: wall-clock Time.

86

43
 Communication Cost Model

How to assess performance?

(1) Computation: Map + Reduce + System Tasks

(2) Communication: Moving key, value pairs

Often dominates computation.
● Connection speeds: 1-10 gigabits per sec;
Ultimate HGDoreaal:d:w5a
0-l1l-5c0lo
gic
gkabT
ytim
[Link] sec
● Even reading from disk to memory typically takes longer than
operating on the data.

87

Communication Cost Model

How to assess performance?

Communication Cost = input size +
(sum of size of all map-to-reducer files)

(2) Communication: Moving key, value pairs

Often dominates computation.
● Connection speeds: 1-10 gigabits per sec;
Ultimate HGDoreaal:d:w5a
0-l1l-5c0lo
gic
gkabT
ytim
[Link] sec
● Even reading from disk to memory typically takes longer than
operating on the data.

88

44
 Communication Cost Model

How to assess performance?

Communication Cost = input size +
(sum of size of all map-to-reducer files)

(2) Communication: Moving key, value pairs

Often dominates computation.
● Connection speeds: 1-10 gigabits per sec;
UltimateHG Doreaal:d:wa-l1l-5c0lo
50 gic
gk
abT
ytie
[Link] sec
● Even reading from disk to memory typically takes longer than
operating on the data.
● Output from reducer ignored because it’s either small (finished
summarizing data) or being passed to another mapreduce job.

89

Communication Cost: Natural Join

R, S: Relations (Tables) R(A, B) ⨝S(B, C)

Communication Cost = input size +

(sum of size of all map-to-reducer files)

DFS Map LocalFS Network Reduce DFS ?

90

45
 Communication Cost: Natural Join

R, S: Relations (Tables) R(A, B) ⨝S(B, C)

Communication Cost = input size +

(sum of size of all map-to-reducer files)

def reduce(k, vs):

r1, r2 = [], []
def map(k, v): for (rel, x) in vs: #separate rs
if k==”R1”: if rel == ‘R’: [Link](x)
(a, b) = v else: [Link](x)
yield (b,(R1,a))
for a in r1: #join as tuple
if k==”R2”:
(b,c) = v for each c in r2:
yield (b,(R2,c)) yield (Rjoin’, (a, k, c)) #k is

91

Communication Cost: Natural Join

R, S: Relations (Tables) R(A, B) ⨝S(B, C)

Communication Cost = input size +

(sum of size of all map-to-reducer files)

= |R1| + |R2| + (|R1| + |R2|)

def reduce(k, vs):
= O(|R1| + |R2|) r1, r2 = [], []
def map(k, v): for (rel, x) in vs: #separate rs
if k==”R1”: if rel == ‘R’: [Link](x)
(a, b) = v else: [Link](x)
yield (b,(R1,a))
for a in r1: #join as tuple
if k==”R2”:
for each c in r2:
(b,c) = v
yield (b,(R2,c)) yield (Rjoin’, (a, k, c)) #k is

92

46
 MapReduce: Final Considerations

● Performance Refinements:
○ Combiners (like word count version 2 but done via reduce)
■ Run reduce right after map from same node before passing to
reduce (MapTask can execute)
■ Reduces communication cost

Requires commutative and associative reducer function.

93

MapReduce: Final Considerations

● Performance Refinements:
○ Combiners (like word count version 2 but done via reduce)
■ Run reduce right after map from same node before passing to
reduce (MapTask can execute)
■ Reduces communication cost

○ Backup tasks (aka speculative tasks)

■ Schedule multiple copies of tasks when close to the end to mitigate
certain nodes running slow.

○ Override partition hash function to organize data

E.g. instead of hash(url) use hash(hostname(url))

94

47