<jupyter_start><jupyter_text>Introduction to Spark Many examples courtesy <NAME>  Outline - Functional programming in Python- Spark's programming model- As many examples as we can get through!  Functional Python Python acquired lambda, reduce, filter and map, courtesy of a Lisp hacker who missed them and submitted working patches. -<NAME>- `map` - `reduce`- `filter`- `lambda`- And more: [itertools](https://docs.python.org/2/library/itertools.html), [pytoolz](https://github.com/pytoolz/toolz/)We will use these concepts (and more) in `Spark`  The `map` abstraction For the category theory inclined: a functor over functions (morphisms)!  Basically an association of functions.<jupyter_code>def square(x):
    return x*x

numbers = [1,2,3]

def map_squares(nums):
    res = []
    for x in nums:
        res.append( square(x) )
    return res

map_squares(numbers)<jupyter_output><empty_output><jupyter_text>or...<jupyter_code>results = map(square, numbers)
results<jupyter_output><empty_output><jupyter_text>For parallel computing in python, `map` is a key abstraction.<jupyter_code>from multiprocessing import Pool
pool = Pool(5)
results = pool.map(square, numbers)
results<jupyter_output><empty_output><jupyter_text>`lambda`  Anonymous function: a function without a name, like inlining<jupyter_code>lambda_square = lambda x: x*x
map(lambda_square, range(10))
map(lambda x: x*x, range(10))
res = map(lambda x: x*x, range(10))<jupyter_output><empty_output><jupyter_text>`reduce`  Apply a function with **two** arguments cumulatively to the container.<jupyter_code>def add_num(x1, x2):
    return x1+x2

print reduce(add_num, res)
print reduce(lambda x,y: x+y, res)<jupyter_output>285<jupyter_text>`filter`  Constructs a new list for items where the applied function is `True`.<jupyter_code>def less_than(x):
    return x>10

filter(less_than, res)
filter(lambda x: x>10, res)<jupyter_output><empty_output><jupyter_text>Spark Programming Model  Everything starts with a `SparkContext`<jupyter_code>import findspark
import os
findspark.init() # you need that before import pyspark in Jupyter notebook

import pyspark
sc = pyspark.SparkContext()<jupyter_output><empty_output><jupyter_text>Create RDDs [RDD Documentation](http://spark.apache.org/docs/latest/api/python/pyspark.htmlpyspark.RDD)The `parallelize` method is a utility for initializing RDDs.- NB: parallelized structure must fit in driver memory!<jupyter_code>import numpy as np

rdd = sc.parallelize(np.arange(20), numSlices=5)<jupyter_output><empty_output><jupyter_text>Transformations and Actions **Transformations** return edges to new vertex in DAG, lazy evaluation, wide and narrow evaluation- `map`, `flatmap`- `reduceByKey`- `filter`- `glom`**Actions** return values- beware of memory limitations!- `collect`- `reduce`- `take`- `count` What does this look like?- `glom`: Return an RDD created by coalescing all elements within each partition into a list.- `collect`: Returns a list from all elements of an RDD.<jupyter_code>for x in rdd.glom().collect():
    print x
rdd = sc.parallelize(np.arange(20), numSlices=10)
for x in rdd.glom().collect():
    print x<jupyter_output>[0, 1]
[2, 3]
[4, 5]
[6, 7]
[8, 9]
[10, 11]
[12, 13]
[14, 15]
[16, 17]
[18, 19]<jupyter_text>`map` and `Flatmap`  Return a new RDD by first applying a function and then flattening the results.<jupyter_code>rdd = sc.parallelize([ [2, 3, 4],[0, 1],[5, 6, 7, 8] ])
rdd.collect()
rdd.map(lambda x: range(len(x))).collect()<jupyter_output><empty_output><jupyter_text>Or I can flatten the results...<jupyter_code>rdd.flatMap(lambda x: range(len(x))).collect()<jupyter_output><empty_output><jupyter_text>Or flatten the original results<jupyter_code>rdd.flatMap(lambda x: x).collect()<jupyter_output><empty_output><jupyter_text>Reduction (Associative operation)<jupyter_code>rdd.flatMap(lambda x: x).reduce(lambda x,y: x+y)
rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 2)])
rdd.collect()
rdd.reduceByKey(lambda x,y: x+y).collect()
rdd = sc.parallelize([("hamlet", 1), ("claudius", 1), ("hamlet", 1)])
rdd.countByKey()<jupyter_output><empty_output><jupyter_text>Reading HDF5 with PySpark  Example courtesy Freeman Lab: https://github.com/freeman-lab/hdf5-and-spark<jupyter_code>import h5py
h5file_path='../data/hdf5_ex.h5'

def readchunk(v):
    chunk = h5py.File(h5file_path, 'r')
    return chunk['/chunked'][v,:]

chunked_array = sc.parallelize(range(0,10)).map(lambda v: readchunk(v))
chunked_array.take(3)<jupyter_output><empty_output><jupyter_text>Now write it to a CSV (from stackoverflow user <NAME>)<jupyter_code>def toCSV(data):
    return ','.join(str(d) for d in data)

lines = chunked_array.map(toCSV).repartition(1)
lines.saveAsTextFile('hdf5_ex.csv')<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13664202745512144, 'fraction_numerical': 0.018585005279831045, 'mean_word_length': 6.411580594679187, 'pattern_counts': {'":': 0, '<': 62, '<?xml version=': 0, '>': 64, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '15', 'dir': 'jupyter-structured-clean-dedup', 'id': '94018', 'license': 'MIT', 'n_tokens_mistral': 1781, 'n_tokens_neox': 1687, 'n_words': 433, 'path': 'exercises/01_introduction.ipynb', 'repo_name': 'milroy/Spark-Meetup', 'size': '4760'}