Data Parallel Computing with Spark
TIPLecture notebook

This is the link to this lecture’s Colab notebook

This notebook has its setup cells merged to simplify the initial run process!

Matrix-vector multiplication I



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

 
import numpy as np

np.random.seed(123)
N = 4 # set matrix and vector dimension

M = np.random.randint(5, size = (N, N))
V = np.random.randint(5, size = N)
P = np.matmul(M, V)

# Print the matrices
print(f"Matrix M:\n {M}")
print(f"Vector V:\n {V}")
print(f"Dot Product P:\n {P}")

mspark = sc.parallelize(M)
print(f"Parallelize M inside Spark Context:\n {mspark.take(4)}")

mspark = sc.parallelize(M).zipWithIndex()
print(f"Parallelize M inside Spark Context with Index:\n {mspark.take(4)}")

mspark = (sc.parallelize(M).zipWithIndex().map(lambda item: item[0].dot(V)))

print(f"MapReduce Dot Product:\n {np.array(mspark.take(4))}")
print(f"Serial Dot Product P for comparison purposes:\n {P}")
TIPScreenshot: Output
Example output
Data initialization
  • Lines 6-7: We initialize matrix M of size N by N, and vector V of size N.
  • Line 8: We carry out a dot product using the builtin function of numpy for later validation.
Parallelizing data in Spark cluster
  • Lines 15-16: Data can be generated on the driver side, then parallelize into RDD objects on the cluster.
    • This will not work well, as we no longer have an indicator of row order. We need to provide some additional information as we parallelize our local data.
  • Lines 18-19: This is possible with zipWithIndex
Direct multiplication
  • Lines 21: We can take each data element of mspark and perform a dot product between the element and Vector V.
  • What happens if Vector V no longer fits into memory?

Matrix-vector multiplication II

In this scenario, V no longer fits into memory and we also need to parallelize V in Spark.



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

 
import numpy as np
np.random.seed(123)
N = 8
M = np.random.randint(5, size = (N, N))
V = np.random.randint(5, size = N)   
P = np.matmul(M, V)
print(f"Matrix M:\n {M}")
print(f"Vector V:\n {V}")
print(f"Dot Product P:\n {P}")

mspark = sc.parallelize(M).zipWithIndex().map(lambda item: (item[1],item[0].tolist()))
print(f"Parallelize M inside Spark Context with Index: {mspark.collect()}")
vspark = sc.parallelize(V).zipWithIndex().map(lambda item: (item[1],item[0].tolist()))
print(f"Parallelize V inside Spark Context with Index: {vspark.collect()}")

def getPos(t):
  res = []
  for idx,x in np.ndenumerate(t[1]):
    res.append((idx[0], (t[0], x.item()))) # (row, (column, cell value))
  return res

def arrayMul(t):
  res = []
  vectorVal = None
  for x in t[1]:
    if type(x) != tuple:
      vectorVal = x
  for x in t[1]:
    if isinstance(x, tuple):
      row_index = x[0]
      matrix_value = x[1]
      res.append((row_index, matrix_value * vectorVal))
  return res

pspark = (mspark.flatMap(getPos)
          .union(vspark)
          .groupByKey()
          .mapValues(list)
          .flatMap(arrayMul)
          .reduceByKey(lambda a, b: a + b))

print(f"MapReduce Dot Product: {pspark.collect()}")
print(f"Serial Dot Product P for comparison purposes:\n {P}")
TIPScreenshot: Output
Example output
Data processing for multiplication
  • Lines 11-14: Both matrix M and vector V are loaded onto Spark
  • Line 11 and line 13: The extra map is so that we flip the indices and use them as keys
Matrix multiplication
  • Because V no longer fits into memory, it can be reasoned that each of these numpy arrays also does not fit in memory.
  • Line 35: Need to extract column identifier from each numpy array and turn them into individual values, similar to the vector’s values with getPos
  • Line 36: Combine individual items of M and V via union,
  • Line 37: Group pair with the same key (position in the resulting vector) together with groupBy,
  • Line 38: Convert data into list type instead of Spark’s iterables,
    • Each element in this frame after groupBy is a key/value pair:
      • The key is the row index
      • The value is a (N+1)-element array:
        • N elements belong to a row of the matrix M, each of this element is a tuple with the key representing the column index and the value represent the corresponding matrix cell value.
      • One remaining element belong to the value of the vector V at that row index position.
  • Line 39: As shown in the previous output, we cannot assume proper order after union, especially for large scale data
    • Need to flatten the final results (list of tupples instead of list of lists of tuples)
  • Line 40: Add elements together with reduceByKey.
CHALLENGEPrinting partial results
  • Create partial results for the transformations between lines 36-40 and print out.

Analyzing text data (not using Spark SQL)

Getting MovieLens data
  • Downloading move review data


1
2

 
!wget https://files.grouplens.org/datasets/movielens/ml-32m.zip
!unzip ml-32m.zip
  • How big of a data set are we talking about?


1

 
!ls -lh ml-32m/
  • You can review the README of this data set.


1
2
3

 
ratings = sc.textFile(working_dir + "/ml-32m/ratings.csv")
print(ratings.take(5))   
ratings.cache()
  • First count, not yet loaded into memory


1
2

 
%%time
ratings.count()

 
CPU times: user 252 ms, sys: 28.2 ms, total: 280 ms
Wall time: 1min 40s
32000205
  • Second count, preloaded


1
2

 
%%time
ratings.count()

 
CPU times: user 147 ms, sys: 19.8 ms, total: 167 ms
Wall time: 58.4 s
32000205
  • Third count, preloaded


1
2

 
%%time
ratings.count()

 
CPU times: user 147 ms, sys: 19.8 ms, total: 167 ms
Wall time: 58.4 s
32000205
CHALLENGEAverage Rating
  • What are the average ratings over the years of each movie?
  • Note: don’t forget to filter!
CHALLENGEAverage Rating with Movie Titles
  • What are the average ratings over the years of each movie?
    • Display the results with movie titles instead of movie IDs
  • Assumption: movies.csv does not fit in memory