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
Piping to external programs

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

In [1]:

import numpy as np
import string

In [2]:

from pyspark import SparkContext
sc = SparkContext('local[*]')

Resources¶

The Spark Programming Guide

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 variable 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.

In [3]:

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

In [4]:

ulysses.take(10)

Out[4]:

['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.

In [5]:

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)

In [6]:

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

In [7]:

counter['circle']

Out[7]:

In [8]:

num_lines.value

Out[8]:

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.

In [9]:

from itertools import count

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

In [10]:

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())

In [11]:

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

Out[11]:

In [12]:

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

Out[12]:

Use SparkContext.broadcast() to create a broadcast variable
Where you would use var, use var.value
The broadcast variabel is sent once to each node and can be re-used

In [13]:

table_bc = sc.broadcast(table)

In [14]:

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())

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.

In [15]:

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

Out[15]:

In [16]:

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

Out[16]:

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

In [17]:

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))

In [18]:

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

Out[18]:

[('uw', 'a'),
 ('iv', 'a'),
 ('cy', 'a'),
 ('to', 'a'),
 ('ea', 'a'),
 ('jc', 'a'),
 ('th', 'a'),
 ('pe', 'a'),
 ('rf', 'a'),
 ('ng', 'a')]

In [19]:

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

In [20]:

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

Out[20]:

[('ro', 'b'),
 ('vf', 'b'),
 ('es', 'b'),
 ('er', 'b'),
 ('kq', 'b'),
 ('gw', 'b'),
 ('jt', 'b'),
 ('my', 'b'),
 ('xx', 'b'),
 ('ui', 'b')]

In [21]:

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

In [22]:

rdd_updated = rdd.join(rdd_new)

In [23]:

rdd_updated.take(10)

Out[23]:

[('sz', ('aaaaaaaaaaaaaaaa', 'bbb')),
 ('sc', ('aaaaaaaaaaaaaa', 'bb')),
 ('kt', ('aaaaaaaaaaaaaaaa', 'b')),
 ('wg', ('aaaaaaaaaaaaaaaaaaa', 'bb')),
 ('vt', ('aaaaaaaaaaaaa', 'b')),
 ('xb', ('aaaaaaaaaaaaaaaaaaa', 'b')),
 ('oa', ('aaaaaaaaaaaaaaa', 'bb')),
 ('uy', ('aaaaaaaaaaaaaaaaaaaaa', 'b')),
 ('gu', ('aaaaaaaaaa', 'bb')),
 ('gb', ('aaaaaaaaaaaaaaaa', 'bb'))]

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.

In [24]:

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

In [25]:

rdd2_updated = rdd2.join(rdd_new)

In [26]:

rdd2_updated.take(10)

Out[26]:

[('sz', ('aaaaaaaaaaaaaaaa', 'bbb')),
 ('tp', ('aaaaaaaaaaaaaaa', 'b')),
 ('sf', ('aaaaaaaaaaaaaaaaaaaaaa', 'bbb')),
 ('qo', ('aaaaaaaaaaaaaaaaaaa', 'bbb')),
 ('nh', ('aaaaaaaaaaaaa', 'b')),
 ('df', ('aaaaaaaaaaaa', 'bb')),
 ('kw', ('aaaaaaaaaaaaaaaaaa', 'bb')),
 ('fo', ('aaaaaaaaaaaaaaa', 'b')),
 ('tl', ('aaaaaaaaaaaaaa', 'bb')),
 ('lh', ('aaaaaaaaaaaaa', 'b'))]

Piping to External Programs¶

Suppose it is more convenient or efficient to write a function in some other language to process data. We can pipe data from Spark to the external program (script) that performs the calculation via standard input and output. The example below shows using a C++ program to calculate the sum of squares for collections of numbers.

In [27]:

%%file foo.cpp

#include <iostream>
#include <sstream>
#include <string>
#include <numeric>
#include <vector>
using namespace std;

double sum_squares(double x, double y) {
  return x + y*y;
};

int main() {

    string s;

    while (cin) {

        getline(cin, s);
        stringstream stream(s);
        vector<double> v;

        while(1) {
            double u;
            stream >> u;
            if(!stream)
                break;
            v.push_back(u);
        }
        if (v.size()) {
            double x = accumulate(v.begin(), v.end(), 0.0, sum_squares);
            cout << x << endl;
        }
    }
}

Overwriting foo.cpp

In [28]:

! g++ foo.cpp -o foo

In [29]:

xs = np.random.random((10, 3))
np.savetxt('numbers.txt', xs)

In [30]:

%%bash

./foo < numbers.txt

In [31]:

%%bash

cat numbers.txt | ./foo

In [32]:

!head numbers.txt

5.270539741683482049e-01 9.538059079198231149e-01 9.705379757246932471e-01
8.224342507879207620e-01 4.863559000065119653e-01 6.055084142317349594e-01
3.776646942097252602e-01 6.458831090447878509e-01 3.890739818068755795e-01
3.894152894179003788e-02 3.690381691456464663e-01 8.589506712912342579e-02
7.596819591747848710e-01 5.785597615102290314e-01 7.884100040999678649e-01
8.717886662425843314e-01 4.836717890004667009e-01 9.547083256957378250e-02
7.107374952186653605e-01 5.218853770211685505e-01 9.323470966394742376e-01
8.793413319051871513e-01 3.959469860304939415e-01 6.483846319494085408e-01
9.579829009525054895e-01 2.739685046015039038e-01 8.817835757073387848e-01
5.315242953832449713e-01 3.254503074377856908e-01 9.383707453566396683e-01

In [33]:

rdd = sc.textFile('numbers.txt')

In [34]:

from pyspark import SparkFiles

def prepare(line):
    """Each line contains numbers separated by a space."""
    return ' '.join(line.split()) + '\n'

# pipe data to external function
func = './foo'
sc.addFile(func)
ss = rdd.map(lambda s: prepare(s)).pipe(SparkFiles.get(func))

In [35]:

np.array(ss.collect(), dtype='float')

Out[35]:

array([ 2.12948 ,  1.27958 ,  0.711174,  0.145084,  1.53344 ,  1.00307 ,
        1.64678 ,  1.35042 ,  1.77033 ,  1.26898 ])

In [36]:

np.sum(xs**2, 1)

Out[36]:

array([ 2.12947556,  1.2795806 ,  0.71117418,  0.14508358,  1.53343841,
        1.00306856,  1.64678324,  1.35041782,  1.77033225,  1.26897563])

Version¶

In [37]:

%load_ext version_information
%version_information pyspark, numpy

Out[37]:

Software	Version
Python	3.5.1 64bit [GCC 4.2.1 (Apple Inc. build 5577)]
IPython	4.0.3
OS	Darwin 15.4.0 x86_64 i386 64bit
pyspark	The 'pyspark' distribution was not found and is required by the application
numpy	1.10.4
Tue Apr 19 13:19:24 2016 EDT

In [ ]:

Using Spark Efficiently¶

Resources¶

Accumulators¶

Event counting¶

Broadcast Variables¶

The Spark Shuffle and Partitioning¶

Using `partitionBy`¶

Piping to External Programs¶

Version¶

Page contents

Previous page

Next page

This Page

Using Spark Efficiently¶

Resources¶

Accumulators¶

Event counting¶

Broadcast Variables¶

The Spark Shuffle and Partitioning¶

Using partitionBy¶

Piping to External Programs¶

Version¶

Using `partitionBy`¶