<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Lecture 7
Step1:  local
Step2:  Lambda functions
Lambda expressions are an easy way to write short functions in Python.
Step3:  Creating RDDS
We saw that we can create RDDs by loading files from disk. We can also create RDDs from Python collections or transforming other RDDs.
Step4:  RDD operations
There are two types of RDD operations in Spark
Step5:  
Step7:  Lazy evaluation
RDDs are evaluated lazily. This means that Spark will not materialize an RDD until it has to perform an action. In the example below, primesRDD is not evaluated until action collect() is performed on it.
Step8:  Persistence
RDDs are ephemeral by default, i.e. there is no guarantee they will remain in memory after they are materialized. If we want them to persist in memory, possibly to query them repeatedly or use them in multiple operations, we can ask Spark to do this by calling persist() on them.
Step9:  If we do not need primesRDD in memory anymore, we can tell Spark to discard it.
Step10:  How long does it take to collect primesRDD? Let's time the operation.
Step11:  When I ran the above on my laptop, it took about more than 10s. That's because Spark had to evaluate primesRDD before performing collect on it.
How long would it take if primesRDD was already in memory?
Step13:  When I ran the above on my laptop, it took about 1s to collect primesRDD - that's almost $10$ times faster compared to when the RDD had to be recomputed from scratch.
Passing functions
When we pass a function as a parameter to an RDD operation, the function can be specified either as a lambda function or as a reference to a function defined elsewhere.
Step14:  Be careful, though
Step15:  The following is a better way to implement the two methods above (get_matches_in_rdd_v1 and get_matches_in_rdd_v2), if we want to avoid sending a SearchFunctions object for computation to the cluster.
Step16:  map and flatmap
Step17:  (Pseudo) set operations
Step18:  reduce
Step19:  aggregate
Step20:  reduceByKey
Step21:  From https
Step22:  (inner) join
Step23:  Accumulators
This example demonstrates how to use accumulators.
The map operations creates an RDD that contains the length of lines in the text file - and while the RDD is materialized, an accumulator keeps track of how many lines are long (longer than $30$ characters).
Step24:  Warning
In the example above, we update the value of an accumulator within a transformation (map). This is not recommended, unless for debugging purposes! The reason is that, if there are failures during the materialization of llengthRDD, some of its partitions will be re-computed, possibly causing the accumulator to double-count some the the long lines.
It is advisable to use accumulators within actions - and particularly with the foreach action, as demonstrated below.
Step25:  Broadcast variable
We use broadcast variables when many operations depend on the same large static object - e.g., a large lookup table that does not change but provides information for other operations. In such cases, we can make a broadcast variable out of the object and thus make sure that the object will be shipped to the cluster only once - and not for each of the operations we'll be using it for.
The example below demonstrates the usage of broadcast variables. In this case, we make a broadcast variable out of a dictionary that represents an address table. The tablke is shipped to cluster nodes only once across multiple operations.
Step26:  Stopping
Call stop() on the SparkContext object to shut it down.
Step27:  Example
Step30:  Example
<ASSISTANT_TASK:>
Python Code:
#On windows
#import findspark
#findspark.init(spark_home="C:/Users/me/software/spark-1.6.3-bin-hadoop2.6/")
import pyspark
import numpy as np # we'll be using numpy for some numeric operations
sc = pyspark.SparkContext(master="local[*]", appName="tour")
sc.stop()
Explanation: Lecture 7: Spark Programming
In what follows, you can find pyspark code for the examples we saw in class.
Many of the examples follow examples found in Learning Spark: Lightning-Fast Big Data Analysis, by Holden Karau, Andy Konwinski, Patrick Wendell, Matei Zaharia, which you can also find at Aalto's library.
Further information also available here: https://spark.apache.org/docs/1.6.0/programming-guide.html
Setup
These instructions should work for Mac and Linux. We'll assume you'll be using python3.
To run the following on your computer, make sure that pyspark is in your PYTHONPATH variable.
You can do that by downloading a zipped file with Spark, extracting it into its own folder (e.g., spark-1.6.0-bin-hadoop2.6/) and then executing the following commands in bash.
export PYSPARK_PYTHON=python3
export SPARK_HOME=/path/to/spark-1.6.0-bin-hadoop2.6/
export PYTHONPATH=$SPARK_HOME/python:$PYTHONPATH
On Windows, the easiest way is to use pyspark is to use Anaconda and then install Jupyter and findspark
End of explanation
# To try the SparkContext with other masters first stop the one that is already running
# sc.stop()
Explanation: local: Run Spark locally with one worker thread (i.e. no parallelism at all).
local[K]: Run Spark locally with K worker threads (ideally, set this to the number of cores on your machine).
local[*]: Run Spark locally with as many worker threads as logical cores on your machine.
spark://HOST:PORT: Connect to the given Spark standalone cluster master. The port must be whichever one your master is configured to use, which is 7077 by default.
mesos://HOST:PORT: Connect to the given Mesos cluster. The port must be whichever one your is configured to use, which is 5050 by default. Or, for a Mesos cluster using ZooKeeper, use mesos://zk://.... To submit with --deploy-mode cluster, the HOST:PORT should be configured to connect to the MesosClusterDispatcher.
yarn: Connect to a YARN cluster in client or cluster mode depending on the value of --deploy-mode. The cluster location will be found based on the HADOOP_CONF_DIR or YARN_CONF_DIR variable.
yarn-client: Equivalent to yarn with --deploy-mode client, which is preferred to yarn-client
yarn-cluster: Equivalent to yarn with --deploy-mode cluster, which is preferred to yarn-cluster
End of explanation
f = lambda line: 'Spark' in line
f("we are learning park")
def f(line):
    return 'Spark' in line
f("we are learning Spark")
Explanation: Lambda functions
Lambda expressions are an easy way to write short functions in Python.
End of explanation
my_list = [0,1,2,3,4,5,6,7,8,9]
data = sc.parallelize(my_list) # create RDD from Python collection
data_squared = data.map(lambda num: num ** 2) # transformation
data_squared.collect()
Explanation: Creating RDDS
We saw that we can create RDDs by loading files from disk. We can also create RDDs from Python collections or transforming other RDDs.
End of explanation
data = sc.parallelize([0,1,2,3,4,5,6,7,8,9]) # creation of RDD
data_squared = data.map(lambda num: num ** 2) # transformation
data_squared.collect() # action
Explanation: RDD operations
There are two types of RDD operations in Spark: transformations and actions. Transfromations create new RDDs from other RDDs. Actions extract information from RDDs and return it to the driver program.
End of explanation
text = sc.textFile("myfile.txt",1) # load data
text = sc.textFile("myfile.txt") # load data
# count only lines that mention "Spark"
spark_lines = text.filter(lambda line: 'Spark' in line)
spark_lines.collect()
Explanation: 
End of explanation
def is_prime(num):
     return True if num is prime, False otherwise
    if num < 1 or num % 1 != 0:
        raise Exception("invalid argument")
    for d in range(2, int(np.sqrt(num) + 1)):
        if num % d == 0:
            return False
    return True
numbersRDD = sc.parallelize(range(1, 1000000)) # creation of RDD
primesRDD = numbersRDD.filter(is_prime) # transformation
# primesRDD has not been materialized until this point
primes = primesRDD.collect() # action
type(primesRDD)
print(primes[0:15])
print(primesRDD.take(15))
Explanation: Lazy evaluation
RDDs are evaluated lazily. This means that Spark will not materialize an RDD until it has to perform an action. In the example below, primesRDD is not evaluated until action collect() is performed on it.
End of explanation
primesRDD.persist()
primesRDD.persist() # we're asking Spark to keep this RDD in memory. Note that cache is the same but as using persist for memory. However, persist allows you to define other types of storage
print("Found", primesRDD.count(), "prime numbers") # first action -- causes primesRDD to be materialized
print("Here are some of them:")
print(primesRDD.take(20)) # second action - RDD is already in memory
Explanation: Persistence
RDDs are ephemeral by default, i.e. there is no guarantee they will remain in memory after they are materialized. If we want them to persist in memory, possibly to query them repeatedly or use them in multiple operations, we can ask Spark to do this by calling persist() on them.
End of explanation
primesRDD.unpersist()
Explanation: If we do not need primesRDD in memory anymore, we can tell Spark to discard it.
End of explanation
%%timeit
primes = primesRDD.collect()
Explanation: How long does it take to collect primesRDD? Let's time the operation.
End of explanation
primesRDD.persist()
%%timeit
primes = primesRDD.collect()
Explanation: When I ran the above on my laptop, it took about more than 10s. That's because Spark had to evaluate primesRDD before performing collect on it.
How long would it take if primesRDD was already in memory?
End of explanation
data = sc.parallelize(range(10))
squares = data.map(lambda x: x**2)
squares.collect()
def f(x):
     return the square of a number
    return x**2
data = sc.parallelize(range(10))
squares = data.map(f)
squares.collect()
Explanation: When I ran the above on my laptop, it took about 1s to collect primesRDD - that's almost $10$ times faster compared to when the RDD had to be recomputed from scratch.
Passing functions
When we pass a function as a parameter to an RDD operation, the function can be specified either as a lambda function or as a reference to a function defined elsewhere.
End of explanation
class SearchFunctions(object):
    def __init__(self, query):
        self.query
        
    def is_match(self, s):
        return self.query in s
    
    def get_matches_in_rdd_v1(self, rdd):
        return rdd.filter(self.is_match) # the function is an object method
    
    def get_matches_in_rdd_v2(self, rdd):
        return rdd.filter(lambda x: self.query in x) # the function references an object field
Explanation: Be careful, though: if the function that you pass as argument to an RDD operation 
* is an object method, or
* references an object field,
then Spark will ship the entire object to the cluster nodes along with the function.
This is demonstrated in the piece of code below.
End of explanation
class SearchFunctions(object):
    def __init__(self, query):
        self.query
        
    def is_match(self, s):
        return self.query in s
    
    def get_matches_in_rdd(self, rdd):
        query = self.query
        return rdd.filter(lambda x: query in x)
Explanation: The following is a better way to implement the two methods above (get_matches_in_rdd_v1 and get_matches_in_rdd_v2), if we want to avoid sending a SearchFunctions object for computation to the cluster.
End of explanation
phrases = sc.parallelize(["hello world", "terve terve", "how are you"])
words_map = phrases.map(lambda phrase: phrase.split(" "))
words_map.collect() # This returns a list of lists
phrases = sc.parallelize(["hello world", "terve terve", "how are you"])
words_flatmap = phrases.flatMap(lambda phrase: phrase.split(" "))
words_flatmap.collect() # This returns a list withe the combined elements of the list
# We can use the flatmap to make a word count
words_flatmap.map(
    lambda x: (x,1)
).reduceByKey(
    lambda x,y: x+y
).collect()
Explanation: map and flatmap
End of explanation
oneRDD = sc.parallelize([1, 1, 1, 2, 3, 3, 4, 4])
oneRDD.persist()
otherRDD = sc.parallelize([1, 4, 4, 7])
otherRDD.persist()
unionRDD = oneRDD.union(otherRDD)
unionRDD.persist()
oneRDD.subtract(otherRDD).collect()
oneRDD.distinct().collect()
oneRDD.intersection(otherRDD).collect() # removes duplicates
oneRDD.cartesian(otherRDD).collect()[:5]
Explanation: (Pseudo) set operations
End of explanation
np.sum([1,43,62,23,52])
data = sc.parallelize([1,43,62,23,52])
data.reduce(lambda x, y: x + y)
data.reduce(lambda x, y: x * y)
data.reduce(lambda x, y: x**2 + y**2) # this does NOT compute the sum of squares of RDD elements
((((1 ** 2 + 43 ** 2) ** 2 + 62 ** 2) **2 + 23 ** 2) **2 + 52 **2)
data.reduce(lambda x, y: np.sqrt(x**2 + y**2)) ** 2
np.sum(np.array([1,43,62,23,52]) ** 2)
Explanation: reduce
End of explanation
help(data.aggregate)
def seq(x,y):
    return x[0] + y, x[1] + 1
def comb(x,y):
    print(x,y,"comb")
    return x[0] + y[0], x[1] + y[1]
data = sc.parallelize([1,43,62,23,52], 1) # Try different levels of paralellism
aggr = data.aggregate(zeroValue = (0,0),
                      seqOp = seq, #
                      combOp = comb)
aggr
aggr[0] / aggr[1] # average value of RDD elements
Explanation: aggregate
End of explanation
help(pairRDD.reduceByKey)
pairRDD = sc.parallelize([('$APPL', 100.64), 
                          ('$APPL', 100.52), 
                          ('$GOOG', 706.2), 
                          ('$AMZN', 552.32), 
                          ('$AMZN', 552.32) ])
pairRDD.reduceByKey(lambda x,y: x + y).collect() # sum of values per key
Explanation: reduceByKey
End of explanation
help(pairRDD.combineByKey)
pairRDD = sc.parallelize([ ('$APPL', 100.64), ('$GOOG', 706.2), ('$AMZN', 552.32), ('$APPL', 100.52), ('$AMZN', 552.32) ])
aggr = pairRDD.combineByKey(createCombiner = lambda x: (x, 1),
                           mergeValue = lambda x,y: (x[0] + y, x[1] + 1),
                           mergeCombiners = lambda x,y: (x[0] + y[0], x[1] + y[1]))
aggr.collect()
aggr.map(lambda x: (x[0], x[1][0]/x[1][1])).collect() # average value per key
Explanation: From https://github.com/vaquarkhan/vk-wiki-notes/wiki/reduceByKey--vs-groupBykey-vs-aggregateByKey-vs-combineByKey
ReduceByKey will aggregate y key before shuffling: 
GroupByKey will shuffle all the value key pairs as the diagrams show: 
combineByKey
End of explanation
help(course_a.join)
course_a = sc.parallelize([ ("Antti", 8), ("Tuukka", 10), ("Leena", 9)])
course_b = sc.parallelize([ ("Leena", 10), ("Tuukka", 10)])
result = course_a.join(course_b)
result.collect()
Explanation: (inner) join
End of explanation
text = sc.textFile("myfile.txt")
long_lines = sc.accumulator(0) # create accumulator
def line_len(line):
    global long_lines # to reference an accumulator, declare it as global variable
    length = len(line)
    if length > 30:
        long_lines += 1 # update the accumulator
    return length
llengthRDD = text.map(line_len)
llengthRDD.count()
long_lines.value # this is how we obtain the value of the accumulator in the driver program
help(long_lines)
long_lines.value # this is how we obtain the value of the accumulator in the driver program
Explanation: Accumulators
This example demonstrates how to use accumulators.
The map operations creates an RDD that contains the length of lines in the text file - and while the RDD is materialized, an accumulator keeps track of how many lines are long (longer than $30$ characters).
End of explanation
text = sc.textFile("myfile.txt")
long_lines_2 = sc.accumulator(0)
def line_len(line):
    global long_lines_2
    length = len(line)
    if length > 30:
        long_lines_2 += 1
text.foreach(line_len)
long_lines_2.value
Explanation: Warning
In the example above, we update the value of an accumulator within a transformation (map). This is not recommended, unless for debugging purposes! The reason is that, if there are failures during the materialization of llengthRDD, some of its partitions will be re-computed, possibly causing the accumulator to double-count some the the long lines.
It is advisable to use accumulators within actions - and particularly with the foreach action, as demonstrated below.
End of explanation
def load_address_table():
    return {"Anu": "Chem. A143", "Karmen": "VTT, 74", "Michael": "OIH, B253.2",
            "Anwar": "T, B103", "Orestis": "T, A341", "Darshan": "T, A325"}
address_table = sc.broadcast(load_address_table())
def find_address(name):
    res = None
    if name in address_table.value:
        res = address_table.value[name]
    return res
people = sc.parallelize(["Anwar", "Michael", "Orestis", "Darshan"])
pairRDD = people.map(lambda name: (name, find_address(name))) # first operation that uses the address table
print(pairRDD.collectAsMap())
other_people = sc.parallelize(["Karmen", "Michael", "Anu"])
pairRDD = other_people.map(lambda name: (name, find_address(name))) # second operation that uses the address table
print(pairRDD.collectAsMap())
Explanation: Broadcast variable
We use broadcast variables when many operations depend on the same large static object - e.g., a large lookup table that does not change but provides information for other operations. In such cases, we can make a broadcast variable out of the object and thus make sure that the object will be shipped to the cluster only once - and not for each of the operations we'll be using it for.
The example below demonstrates the usage of broadcast variables. In this case, we make a broadcast variable out of a dictionary that represents an address table. The tablke is shipped to cluster nodes only once across multiple operations.
End of explanation
sc.stop()
Explanation: Stopping
Call stop() on the SparkContext object to shut it down.
End of explanation
import random
NUM_SAMPLES = 10000000
def inside(p):
    x, y = random.random(), random.random()
    return x*x + y*y < 1
count = sc.parallelize(range(0, NUM_SAMPLES)).filter(inside).count()
print("Pi is roughly %f" % (4.0 * count / NUM_SAMPLES))
Explanation: Example: Estimating Pi
End of explanation
# Details of the algorithm can be found here: http://www.cs.princeton.edu/~chazelle/courses/BIB/pagerank.htm
iterations = 5
def computeContribs(urls, rank):
    Calculates URL contributions to the rank of other URLs.
    num_urls = len(urls)
    for url in urls:
        yield (url, rank / num_urls)
def parseNeighbors(urls):
    Parses a urls pair string into urls pair.
    parts = urls.split(',')
    return parts[0], parts[1]
# Read the lines
lines = sc.textFile("higgs-mention_network.txt").persist()
lines.collect()
# Loads all URLs from input file and initialize their neighbors.
links = lines.map(lambda urls: parseNeighbors(urls)).distinct().groupByKey().cache()
links.collect()
# Loads all URLs with other URL(s) link to from input file and initialize ranks of them to one.
ranks = links.map(lambda url_neighbors: (url_neighbors[0], 1.0))
ranks.collect()
# Calculates and updates URL ranks continuously using PageRank algorithm.
for iteration in range(iterations):
    # Calculates URL contributions to the rank of other URLs.
    contribs = links.join(ranks).flatMap(
        lambda url_urls_rank: computeContribs(url_urls_rank[1][0], url_urls_rank[1][1]))
    # Re-calculates URL ranks based on neighbor contributions.
    ranks = contribs.reduceByKey(lambda x,y: x+y).mapValues(lambda rank: rank * 0.85 + 0.15)
# Collects all URL ranks and dump them to console.
for (link, rank) in ranks.collect():
    print("%s has rank: %s." % (link, rank))
Explanation: Example: Computing PageRank
End of explanation
<END_TASK>