Using Spark Efficiently

Focus in this lecture is on Spark constructs that can make your programs more efficient. In general, this means minimizing the amount of data transfer across nodes, since this is usually the bottleneck for big data analysis problems.

• Shared variables

  • Accumulators

  • Broadcast variables

• DataFrames

• Partitioning and the Spark shuffle

Spark tuning and optimization is complicated - this tutorial only touches on some of the basic concepts.

Don’t forget the otehr areas of optimizaiton shown in previous notebooks:

• Use DataFrmaes rather than RDDs

• Use pyspark.sql.functions rather than a Python UDF

• If you use a UDF, see if you can use a vectorized UDF

from pyspark import SparkContext

sc =  SparkContext.getOrCreate()

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .master("local")
    .appName("BIOS-823")
    .config("spark.executor.cores", 4)
    .getOrCreate()
)

import numpy as np
import string

Resources

The Spark Programming Guide

Shared variables

The second abstraction in Spark are shared variabels, consisting of accumulators and broadcast variables.

Source: https://jaceklaskowski.gitbooks.io/mastering-apache-spark/content/images/sparkcontext-broadcast-executors.png

Accumulators

Spark functions such as map can use variables defined in the driver program, but they make local copies of the variable that are not passed back to the driver program. Accumulators are shared variables that allow the aggregation of results from workers back to the driver program, for example, as an event counter. Suppose we want to count the number of rows of data with missing information. The most efficient way is to use an accumulator.

ulysses = sc.textFile('data/texts/Ulysses.txt')

ulysses.take(10)

['The Project Gutenberg EBook of Ulysses, by James Joyce',
 '',
 'This eBook is for the use of anyone anywhere at no cost and with',
 'almost no restrictions whatsoever.  You may copy it, give it away or',
 're-use it under the terms of the Project Gutenberg License included',
 'with this eBook or online at www.gutenberg.org',
 '',
 '',
 'Title: Ulysses',
 '']

Event counting

Notice that we have some empty lines. We want to count the number of non-empty lines.

num_lines = sc.accumulator(0)

def tokenize(line):
    table = dict.fromkeys(map(ord, string.punctuation))
    return line.translate(table).lower().strip().split()

def tokenize_count(line):
    global num_lines

    if line:
        num_lines += 1

    return tokenize(line)

counter = ulysses.flatMap(lambda line: tokenize_count(line)).countByValue()

counter['circle']

20

num_lines.value

25396

Broadcast Variables

Sometimes we need to send a large read only variable to all workers. For example, we might want to share a large feature matrix to all workers as a part of a machine learning application. This same variable will be sent separately for each parallel operation unless you use a broadcast variable. Also, the default variable passing mechanism is optimized for small variables and can be slow when the variable is large.

from itertools import count

table = dict(zip(string.ascii_letters, count()))

def weight_first(line, table):
    words = tokenize(line)
    return sum(table.get(word[0], 0) for word in words if word.isalpha())

def weight_last(line, table):
    words = tokenize(line)
    return sum(table.get(word[-1], 0) for word in words if word.isalpha())

The dictionary table is sent out twice to worker nodes, one for each call

ulysses.map(lambda line: weight_first(line, table)).sum()

2941855

ulysses.map(lambda line: weight_last(line, table)).sum()

2995994

Converting to use broadast variables is simple and more efficient

• Use SparkContext.broadcast() to create a broadcast variable

• Where you would use var, use var.value

• The broadcast variable is sent once to each node and can be re-used

table_bc = sc.broadcast(table)

def weight_first_bc(line, table):
    words = tokenize(line)
    return sum(table.value.get(word[0], 0) for word in words if word.isalpha())

def weight_last_bc(line, table):
    words = tokenize(line)
    return sum(table.value.get(word[-1], 0) for word in words if word.isalpha())

table_bc is sent to nodes only once.

Although it looks like table_bc is being passed to each function, all that is passed is a path to the table. The worker checks if the path has been cached and uses the cache instead of loading from the path.

ulysses.map(lambda line: weight_first_bc(line, table_bc)).sum()

2941855

ulysses.map(lambda line: weight_last_bc(line, table_bc)).sum()

2995994

The Spark Shuffle and Partitioning

Some events trigger the redistribution of data across partitions, and involves the (expensive) copying of data across executors and machines. This is known as the shuffle. For example, if we do a reduceByKey operation on key-value pair RDD, Spark needs to collect all pairs with the same key in the same partition to do the reduction.

For key-value RDDs, you have some control over the partitioning of the RDDs. In particular, you can ask Spark to partition a set of keys so that they are guaranteed to appear together on some node. This can minimize a lot of data transfer. For example, suppose you have a large key-value RDD consisting of user_name: comments from a web user community. Every night, you want to update with new user comments with a join operation

def fake_data(n, val):
    users = list(map(''.join, np.random.choice(list(string.ascii_lowercase), (n,2))))
    comments = [val]*n
    return tuple(zip(users, comments))

data = fake_data(10000, 'a')
list(data)[:10]

[('vy', 'a'),
 ('vv', 'a'),
 ('us', 'a'),
 ('cx', 'a'),
 ('fq', 'a'),
 ('qo', 'a'),
 ('mr', 'a'),
 ('fs', 'a'),
 ('hd', 'a'),
 ('zc', 'a')]

rdd = sc.parallelize(data).reduceByKey(lambda x, y: x+y)

new_data = fake_data(1000,  'b')
list(new_data)[:10]

[('im', 'b'),
 ('it', 'b'),
 ('pv', 'b'),
 ('xj', 'b'),
 ('ye', 'b'),
 ('xa', 'b'),
 ('fv', 'b'),
 ('nl', 'b'),
 ('nl', 'b'),
 ('vo', 'b')]

rdd_new = sc.parallelize(new_data).reduceByKey(lambda x, y: x+y).cache()

rdd_updated = rdd.join(rdd_new)

rdd_updated.take(10)

[('hd', ('aaaaaaaaaaaaaaa', 'bbbbbb')),
 ('gx', ('aaaaaaaaaaaaaaaaaa', 'bb')),
 ('xj', ('aaaaaaaaaaaaa', 'bbb')),
 ('oa', ('aaaaaaaaaaaaaa', 'bb')),
 ('ow', ('aaaaaaaaaaaaaaaaa', 'bb')),
 ('ej', ('aaaaaaaaaaaaaaa', 'bbb')),
 ('im', ('aaaaaaaaaaa', 'bb')),
 ('gu', ('aaaaaaaaaaa', 'bb')),
 ('cs', ('aaaaaaaaaaaaaaaaaaaaa', 'b')),
 ('qy', ('aaaaaaaaaaaaaaa', 'bbb'))]

Using partitionBy

The join operation will hash all the keys of both rdd and rdd_nerw, sending keys with the same hashes to the same node for the actual join operation. There is a lot of unnecessary data transfer. Since rdd is a much larger data set than rdd_new, we can instead fix the partitioning of rdd and just transfer the keys of rdd_new. This is done by rdd.partitionBy(numPartitions) where numPartitions should be at least twice the number of cores.

From the R docs for partitionBy

This function operates on RDDs where every element is of the form list(K, V) or c(K, V). For each element of this RDD, the partitioner is used to compute a hash function and the RDD is partitioned using this hash value.

In other words, which parittion a data element is sent to depends on the key value.

rdd_A = sc.parallelize([1, 2, 3, 4, 2, 4, 1]).map(lambda x: (x, x))
for item in rdd_A.partitionBy(4).glom().collect():
    print(item)

[(4, 4), (4, 4)]
[(1, 1), (1, 1)]
[(2, 2), (2, 2)]
[(3, 3)]

rdd_B = sc.parallelize([(4,'a'), (1,'b'), (2, 'c'), (3, 'd'), (4,'e'), (1, 'f')])

for item in rdd_B.glom().collect():
    print(item)

[]
[(4, 'a')]
[]
[(1, 'b')]
[]
[(2, 'c')]
[]
[(3, 'd')]
[]
[(4, 'e')]
[]
[(1, 'f')]

rdd_comb = rdd_A.join(rdd_B).glom()

Note: See how all the items from rdd_B have been transferred to the partitions created by rdd_A, but the items from rdd_A have not moved. If rdd_A is much larger than rdd_B then this minimizes the amount of data transfer.

for item in rdd_comb.collect():
    print(item)

[]
[(1, (1, 'b')), (1, (1, 'f')), (1, (1, 'b')), (1, (1, 'f'))]
[(2, (2, 'c')), (2, (2, 'c'))]
[(3, (3, 'd'))]
[(4, (4, 'a')), (4, (4, 'e')), (4, (4, 'a')), (4, (4, 'e'))]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]
[]

Applyin to our word counts

rdd2 = sc.parallelize(data).reduceByKey(lambda x, y: x+y)
rdd2 = rdd2.partitionBy(10).cache()

rdd2_updated = rdd2.join(rdd_new)

rdd2_updated.take(10)

[('mu', ('aaaaaaaaaaaaaaaa', 'bb')),
 ('sl', ('aaaaaaaaaaaaaaaa', 'b')),
 ('kr', ('aaaaaaaaa', 'bb')),
 ('su', ('aaaaaaaaaaaaaaaaaa', 'bb')),
 ('qs', ('aaaaaaaaaaa', 'bb')),
 ('hy', ('aaaaaaaaaaaaaaa', 'bb')),
 ('gi', ('aaaaaaaaaaaaaaaaaa', 'bbbb')),
 ('cb', ('aaaaaaaaaaaaaaa', 'bb')),
 ('yt', ('aaaaaaaaaaaa', 'b')),
 ('qd', ('aaaaaaaaaaaaaaaaaaaa', 'bbbb'))]

spark.stop()