<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 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.
Step1:  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 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.
Step2:  Event counting
Notice that we have some empty lines. We want to count the number of non-empty lines.
Step3:  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.
Step4:  The dictionary table is sent out twice to worker nodes, one for each call
Step5:  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
Step6:  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.
Step7:  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
Step8:  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.
Step9:  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.
Step10:  Feed data via re-direction
Step12:  Feed data via piping
Step13:  Version
<ASSISTANT_TASK:>
Python Code:
import numpy as np
import string
from pyspark import SparkContext
sc = SparkContext('local[*]')
Explanation: 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.
End of explanation
ulysses = sc.textFile('data/Ulysses.txt')
ulysses.take(10)
Explanation: 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 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.
End of explanation
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']
num_lines.value
Explanation: Event counting
Notice that we have some empty lines. We want to count the number of non-empty lines.
End of explanation
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())
Explanation: 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.
End of explanation
ulysses.map(lambda line: weight_first(line, table)).sum()
ulysses.map(lambda line: weight_last(line, table)).sum()
Explanation: The dictionary table is sent out twice to worker nodes, one for each call
End of explanation
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())
Explanation: 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
End of explanation
ulysses.map(lambda line: weight_first_bc(line, table_bc)).sum()
ulysses.map(lambda line: weight_last_bc(line, table_bc)).sum()
Explanation: 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.
End of explanation
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]
rdd = sc.parallelize(data).reduceByKey(lambda x, y: x+y)
new_data = fake_data(1000,  'b')
list(new_data)[:10]
rdd_new = sc.parallelize(new_data).reduceByKey(lambda x, y: x+y).cache()
rdd_updated = rdd.join(rdd_new)
rdd_updated.take(10)
Explanation: 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
End of explanation
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)
Explanation: 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.
End of explanation
%%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;
        }
    }
}
! g++ foo.cpp -o foo
xs = np.random.random((10, 3))
np.savetxt('numbers.txt', xs)
Explanation: 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.
End of explanation
%%bash
./foo < numbers.txt
Explanation: Feed data via re-direction
End of explanation
%%bash
cat numbers.txt | ./foo
!head numbers.txt
rdd = sc.textFile('numbers.txt')
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))
np.array(ss.collect(), dtype='float')
np.sum(xs**2, 1)
Explanation: Feed data via piping
End of explanation
%load_ext version_information
%version_information pyspark, numpy
Explanation: Version
End of explanation
<END_TASK>