10-605 / 10-805�Map-Reduce and Spark�Sample Workflows

Machine Learning from Large Datasets

Recap 1/2

• Administrivia History of big data for ML
• To learn something complex you need lots and lots of data – the more the better!
• Deep learning in general, and LLMs in particular, need big data
• Big models are not enough
• Working with big data means understanding how to do ML efficiently
• Starting with standard complexity analysis and the cost of operations
• Best case for i/o bound tasks is sequential access to data (from disk or from a network)

Recap 2/2

• Cost of operations
• read from disk sequentially
• often data size > RAM
• Learning as counting
• density estimation
• naïve Bayes
• words are independent of each other given the class
• Distributed processing for big data
• embarrassingly parallel tasks
• map-reduce
• abstract operations: map, flatMap, filter, reduceByKey, ….
• Spark / pyspark

​

Recap 3/2: Spark WordCount

4

Recap 3/2: Spark WordCount

5

Recap 3/2: Spark WordCount

6

1. Tokenization and converting to messages is done in parallel over each partition of the RDD lines
2. reduceByKey does a shuffle-sort, doing a pairwise reduce of each “run” of messages with the same “key”

Today’s Plan

• Where did Map-Reduce come from?
• wordcount with Unix tools
• A simple implementation
• From Map-Reduce to Spark
• Example Workflows in Spark

​

7

MAP-REDUCE: AN ORIGIN STORY

8

How hacks to handle large datasets with small memory evolved into Hadoop, Spark, Flume, …

9

c. 1994

Some examples

tr -sc '[:alpha:]' '\n’

< data/brown_nolines.txt

| sort

| uniq -c

| sort -nr

​

10

translate non-letters (-c) to newline and ‘squeeze’ runs of newlines to a single one

The

Fulton

County

Grand

Jury

said

Friday

an

investigation

of

…

sort words

dedup and count

sort in descending (-r) numeric order

62690 the

36043 of

27952 and

25725 to

22000 a

19581 in

10328 that

9969 is

9770 was

8833 for

11

Unix poetry

• Line-by-line processing of a file
• tr (translate): e.g. line of text 🡪 list of words
• awk: line of text 🡪 simple program output
• e.g. bigram + statistics 🡪 MI score
• …
• Filtering
• e.g., discard infrequent bigrams with awk
• Sorting
• and then counting runs of unique values

​

12

Line-by-line processing, filtering is memory efficient

​

So is sorting!

13

Merge sort

14

Merge sort

15

Alternative visualization

Unix Sort

16

• Load as much as you can into memory (S) and do an in-memory sort [usually quicksort].

​

• If you have more to do, then spill this sorted buffer out on to disk, and get more data…

​

• Finally, merge your spill buffers.

Back to Map-Reduce…

17

map takes one input x and outputs any number of outputs

reduce takes a key (word) and an associated Iterator over associated values and outputs some aggregation of the values

Recap 3/2: Spark WordCount

18

A Bad Implementation

19

bash-3.2$ head -1 ../data/brown_nolines.txt

​

The Fulton County Grand Jury said Friday an investigation of Atlanta's recent primary election produced "no evidence" that any irregularities took place.

​

bash-3.2$ py -m fire wc_nano.py WordCount \ map_reduce

--src ../data/brown_nolines.txt --dst wc.txt

​

bash-3.2$ head -5 wc.txt

('the', 69969)

('fulton', 17)

('county', 160)

('grand', 52)

('jury', 68)

wc_nano.py

hz_nano.py

python fire –m <script> <class> <method> --arg1 val1 …

Constant-Memory Word Count in Map-Reduce

20

bash-3.2$ head -1 ../data/brown_nolines.txt

​

The Fulton County Grand Jury said Friday an investigation of Atlanta's recent primary election produced "no evidence" that any irregularities took place.

​

bash-3.2$ py -m fire wc_nano.py WordCount \ map_reduce

--src ../data/brown_nolines.txt --dst wc.txt

​

bash-3.2$ head -5 wc.txt

('the', 69969)

('fulton', 17)

('county', 160)

('grand', 52)

('jury', 68)

Constant-Memory Word Count in Map-Reduce

21

bash-3.2$ head -1 ../data/brown_nolines.txt

​

The Fulton County Grand Jury said Friday an investigation of Atlanta's recent primary election produced "no evidence" that any irregularities took place.

​

bash-3.2$ py -m fire wc_nano.py WordCount \ map_reduce

--src ../data/brown_nolines.txt --dst wc.txt

​

bash-3.2$ head -5 wc.txt

('the', 69969)

('fulton', 17)

('county', 160)

('grand', 52)

('jury', 68)

Python generators 101

​

Each time g.__next__() is called:

• run g until it ‘yields’ a value
• save the state of g

​

Constant memory (state of g)

pair_generator() produces

(‘0’, 1), … (‘aaa’, 1), … (‘aaron’, 1), …

ReduceReady(pair_generator()) returns

​

(‘aaron’, <generator_aaron>)

​

where <generator_aaron>) produces: 1, 1 …

22

or you could use itertools.group_by

Constant-Memory Word Count in Map-Reduce

23

bash-3.2$ head -1 ../data/brown_nolines.txt

​

The Fulton County Grand Jury said Friday an investigation of Atlanta's recent primary election produced "no evidence" that any irregularities took place.

​

bash-3.2$ py -m fire wc_micro.py WordCount \ map_reduce

--src ../data/brown_nolines.txt --dst wc.txt

​

bash-3.2$ head -5 wc.txt

('0', 108)

('00', 33)

('000', 427)

('001', 2)

('002', 1)

Parallel constant-memory wordcount

24

• doc 1
• doc 2
• doc 3
• ….

Counting logic

Pass 1

Sort

• ctr[w1] += 1
• ctr[w1] += 1
• ….

Logic to process messages

Pass 2

Sorting

“ctr[w1] +=1”

“ctr[w2] +=1”

…

Parallel constant-memory wordcount

25

• doc 1
• doc 2
• doc 3
• ….

Counting logic

“ctr[x] +=d”

Pass 1: Mapper

Sort

• ctr[x1] += d1
• ctr[x1] += d2
• ….

Logic to combine counter updates

Pass 2: Reducer

Spill 1

Spill 2

Spill 3

…

Merge Spill Files

Observation: you can “reduce” in parallel (correctly) if no sort key is split across machines

26

• doc 1
• doc 2
• doc 3
• ….

Counting logic

“ctr[x] +=d”

Mapper Machine

Sort

Spill 1

Spill 2

Spill 3

…

Spill 4

Sort needs to partition data appropriately

(into multiple “shards”)

Parallel constant-memory wordcount

27

• doc 1
• doc 2
• doc 3
• ….

Counting logic

“ctr[x] +=d”

Mapper Machine

Sort

Spill 1

Spill 2

Spill 3

…

Spill 4

Sort needs to partition data appropriately, i.e., by the sort key

Parallel constant-memory wordcount

28

• doc 1
• doc 2
• ….

Counting logic

Mapper Machine 1

Partition/Sort

Spill 1

Spill 2

Spill 3

…

• doc 1
• doc 2
• ….

Counting logic

Mapper Machine 2

Partition/Sort

Spill 1

Spill 2

Spill 3

…

Spill n

Same holds for mapper machines: you can count in parallel if no sort key is split across multiple reducer machines

Simple way to partition:

hash the sort key

29

• doc 1
• doc 2
• ….

Counting logic

Mapper Machine 1

Partition/Sort

Spill 1

Spill 2

Spill 3

…

• doc 1
• doc 2
• ….

Counting logic

Mapper Machine 2

Partition/Sort

Spill 1

Spill 2

Spill 3

…

Spill n

30

• doc 1
• doc 2
• ….

Counting logic

Mapper Machine 1

Partition/Sort

Spill 1

Spill 2

Spill 3

…

• doc 1
• doc 2
• ….

Counting logic

Mapper Machine 2

Partition/Sort

Spill 1

Spill 2

Spill 3

…

Spill n

Mapper/counter machines run the “Map phase” of map-reduce

​

• Input different subsets of the total inputs
• Output (sort key,value) pairs

​

• The (key,value) pairs are partitioned based on the key
• Pairs from each partition will be sent to a different reducer machine.

​

31

The shuffle/sort phrase:

​

• (key,value) pairs from each partition are sent to the right reducer
• The reducer will sort the pairs together to get all the pairs with the same key together.

​

The reduce phase:

• Each reducer will scan through the sorted key-value pairs.

32

Keys for partitioning and sorting are common elements of many map-reduce frameworks.

​

Ideally the partitions will be similar size – but you also need to assign keys to partitions quickly.

​

Random hashing is often fine, but not when some keys have lots of messages and some have few.

Map-Reduce is Distributed �Map, Shuffle-Sort, Reduce

33

• doc 1
• doc 2
• doc 3
• ….

Counting logic

Map Process 1

Partition/Sort

Spill 1

Spill 2

Spill 3

…

• doc 1
• doc 2
• doc 3
• ….

Counting logic

Map Process 2

Partition/Sort

Spill 1

Spill 2

Spill 3

…

• C[x1] += D1
• C[x1] += D2
• ….

Logic to combine counter updates

Reducer 1

Merge Spill Files

• C[x2] += D1
• C[x2] += D2
• ….

combine counter updates

Reducer 2

Merge Spill Files

Spill n

Distributed Shuffle-Sort

Key ideas in distributed map reduce

34

Key ideas in distributed map reduce

35

Key ideas in distributed map reduce

36

Today’s Plan

• Where did Map-Reduce come from?
• wordcount with Unix tools
• A simple implementation
• From Map-Reduce to Spark
• Example Workflows in Spark

​

37

More About Spark

38

Spark is MapReduce++

• Data abstraction
• Sharded files 🡪 Resilient Distributed Datasets
• RDDs are immutable
• They are partitioned / sharded
• You access them with high level operations that can be implemented by short Map-Reduce pipelines
• You can mark RDDs as ‘persistent’ (the data is saved for debugging, reuse, etc) and RDDs can persist on disk or in memory
• Memory-cached RDDs use a lot of cluster memory but are very useful
• Two types of operations on RDDs
• Transformations – lazy, return a ‘plan’ to make some change
• Actions – eager, usually force execution of a map-reduce pipeline, and return some Python object

​

39

Many Things Are Built On Spark

• DB-like data processing
• SparkSQL
• Spark DataFrames
• ML libraries
• MLlib
• Graph libraries
• GraphX
• …

​

• We’re mostly not covering these here!

40

Why is Caching in Memory Important?

41

Spark examples: logistic regression

​

• Model:

​

• Learning algorithm:
• define the “loss” of the model on the data
• start with random w and iteratively adjust it to reduce loss
• by gradient descent

Why is Caching in Memory Important?

43

Spark examples: batch logistic regression

44

w is defined outside the lambda function, but used inside it

So: python builds a closure – code including the current value of w – and Spark ships it off to each worker. So w is copied, and must be read-only.

Spark examples: batch logistic regression

45

dataset of points is cached in cluster memory to reduce i/o

46

• point 1
• point 2
• point 3
• ….

Gradient logic

Map Process 1

• point 7,356,701
• point 7,356,702
• point 7,356,703
• ….

Gradient logic

Map Process 2

Partition/Sort

• d/dw point1
• …
• ….

add pointwise gradients

Reducer 1

​

• d/dw point 7..001
• ….

add pointwise gradients

Reducer 2

Distributed Shuffle-Sort

gradient

w

Why is Caching in Memory Important?

• Without caching the training data, you need to load it all the data from disk in each iteration.
• With caching, communication/disk storage is only need for the weights.
• ML is all about iterative optimization.

47

Why Transformations and Actions?

• For large data it’s important to have an explicit representation of the “code” you want to run
• For optimization
• e.g. a sequence of .map() transformations
• For distribution to workers
• For recovery from errors
• …
• Spark’s transformations are like “code”
• This happens over and over in different contexts
• Databases have query languages and query optimizers
• Tensorflow and PyTorch have “computation graphs”
• …
• Language-like packages in Python are not different

48

Why Transformations and Actions?

• Actions give you a way to use Spark interactively
• Use them when you debug
• Avoid them as intermediate steps in a “production” pipeline
• They are bottlenecks!

​

49

Cost of Transformations

• Some transformations can run as a single cheap “mapper” process
• Each partition is transformed independently, so you don’t need to move data around
• The data keys are unchanged, so you don’t need to re-partition the data
• Examples:
• filter(f), sample(…), mapValues(f), …
• map(f, preservesPartioning=True)
• foreach(f) # only runs f, doesn’t update RDD

50

Cost of Transformations

• Some transformations require an expensive shuffle-sort
• reduceByKey, join, intersection, subtract, groupBy, sortBy,…
• partitionByKey, repartition, ..

​

• It’s not always obvious when a shuffle-sort will be needed
• But if you understand the map-reduce layer you can make a good guess

51

Communication without RDDs

• Broadcast variables can be used to share constants, config information, and small data structures to all workers.
• Like closures in a function call, they are copied to every worker
• Worker-driven hanges to a broadcast variable are not broadcast back
• Helpful for implementing a “map-side” join

​

b = sc.broadcast(obj)

if b.value > 10:

…

52

Communication without RDDs

• Accumulators are shared counters
• Generally used for monitoring progress
• You don’t know the order in which accumulators will be incremented
• You can define your own accumulator classes
• Workers can’t read the accumulator’s values

53

Communication without RDDs

• Closures sent to workers as an argument of a transformation or action

54

Communication without RDDs

• Closures sent to workers as an argument of a transformation or action
• note: lambda’s in Python are sometimes surprising!
• be careful and/or be explicit!

55

Communication without RDDs

• Closures sent to workers as an argument of a transformation or action
• Q: how does Spark determine what values to package up with the code that it sends to workers?

​

• Good practice: don’t make this too complicated

56

A CONSTANT-MEMORY VERSION OF SPARK

57

58

59

PageRank

60

Google’s PageRank

61

web site xxx

web site yyyy

web site a b c d e f g

web

site

pdq pdq ..

web site yyyy

web site a b c d e f g

web site xxx

Inlinks are “good” (recommendations)

Inlinks from a “good” site are better than inlinks from a “bad” site

but inlinks from sites with many outlinks are not a “good”...

“Good “ and “bad” are relative.

web site xxx

Google’s PageRank

62

web site xxx

web site yyyy

web site a b c d e f g

web

site

pdq pdq ..

web site yyyy

web site a b c d e f g

web site xxx

Imagine a “pagehopper” that always either

• follows a random link, or
• jumps to random page

Google’s PageRank�(Brin & Page, http://www-db.stanford.edu/~backrub/google.html)

63

web site xxx

web site yyyy

web site a b c d e f g

web

site

pdq pdq ..

web site yyyy

web site a b c d e f g

web site xxx

Imagine a “pagehopper” that always either

• follows a random link, or
• jumps to random page

PageRank ranks pages by the amount of time the pagehopper spends on a page:

• or, if there were many pagehoppers, PageRank is the expected “crowd size”

PageRank in Memory

• Let u = (1/N, …, 1/N)
• dimension = #nodes N
• Let A = adjacency matrix: [a_ij=1 ⬄ i links to j]
• Let W = [w_ij = a_ij/outdegree(i)]
• w_ij is probability of jump from i to j
• Let v⁰ = (1,1,….,1)
• or anything else you want
• Repeat until converged:
• Let v^t+1 = cu + (1-c) v^tW
• c is probability of jumping “anywhere randomly”

64

PageRank in Spark

65

A Slow Implementation

nodes: list of string ids

outlinks_by_node: RDD of (src, [out1, .. ])

v⁰ = (1,1,….,1)

(1-c) v^tW

​

v^t+1 = cu + (1-c) v^tW

​

cu

w_ij is probability of jump from i to j

​

(1-c) v^tW

after sum

​

A Slow Implementation

10 iterations on small graph locally is 10min

​

5 iterations is 30sec

A Slow Implementation

Fixing A Slow Implementation

Keeping v in memory

• Convert values to “combiner” (accumulator type)
• Merge value with accumulator
• Separately merge two accumulators
• Leads to important optimizations because combine can happen on mapper side

Explicit closure

COMPARING COUNTS IN TWO CORPORA

Classic PySpark examples: wordcount

72

Naïve Bayes:

Comparing Bigrams

Problem: many of the common bigrams don’t seem to have meaning as a phrase….

A Language Model Approach to Keyphrase Extraction

Scoring bigrams as phrases

As a phrase “x y”

As a random co-occurrence of x and y

Comparing Phrases

Filtered for high “phrasiness”

Comparing bigrams and compute filter by phrasiness

The pipeline – basic counts

The pipeline w/o filtering

Filtering phrases

Joins change shape of the data

This gets awkward…

Filtering phrases

An example from Ron Bekkerman�k-means in Map-Reduce�(background)

83

Expectation-Maximization

84

EM is a kind of coordinate ascent:

1. Split variables w to optimize into two sets: x,y
2. Fix one set (x) and optimize the other (y)
3. Then fix y and optimize x
4. Then repeat…

​

​

​

​

​

​

Expectation-Maximization

85

In EM there is typically

• instances X
• a “soft assignment” of each x_i to a “cluster”, z_i
• z_i,j = Pr(x_i in cluster j)
• parameters w

​

Then do coordinate descent on w, Z to maximize

​

–log Pr(X| w, Z)

​

An example from Ron Bekkerman�k-means in Map-Reduce

86

Example: k-means clustering

• An EM-like algorithm:
• Initialize k cluster centroids
• E-step: associate each data instance with the closest centroid
• Find “expected values” of cluster assignments given the data and centroids
• M-step: recalculate centroids as an average of the associated data instances
• Find new centroids that “maximize that expectation”
• And repeat…

87

k-means Clustering

88

centroids

Example: k-means clustering

• An EM-like algorithm:
• Initialize k cluster centroids
• E-step: associate each data instance with the closest centroid
• M-step: recalculate centroids as an average of the associated data instances

89

Parallelizing k-means

90

Parallelizing k-means

91

Parallelizing k-means

92

k-means on MapReduce

• Mappers read data partitions + centroids
• Mappers assign data instances to clusters
• Mappers compute new local centroids and local cluster sizes
• Reducers aggregate local centroids (weighted by local cluster sizes) into new global centroids
• Reducers write the new centroids

93

Questions

• Mappers read data portions and centroids
• Mappers assign data instances to clusters
• Mappers compute new local centroids and local cluster sizes
• Reducers aggregate local centroids (weighted by local cluster sizes) into new global centroids
• Reducers write the new centroids

94

• What should be in memory and what should be in an RDD?
• How many Shuffle-Sorts are needed for one pass?
• Assume we can map through documents

Questions

• Mappers read data portions and centroids
• Mappers assign data instances to clusters
• Mappers compute new local centroids and local cluster sizes
• Reducers aggregate local centroids (weighted by local cluster sizes) into new global centroids
• Reducers write the new centroids

95

Sketch:

• Map only: With all centroids in mapper memory, assign each doc to a centroid
• outputting docid 🡪 centroid index
• (Shuffle-sort: Count number of docs per centroid)
• Map-Reduce: join docs to centroids
• Map-Reduce: average doc weights for each centroid