<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 <a>
<img align=left src="files/images/pyspark-page1.svg" width=500 height=250 />
</a>
DataFrame API
GitHub
related blog post
<a>
<img align=left src="files/images/pyspark-page2.svg" width=500 height=500>
</a>
Click on a picture to view pyspark docs
Step1:  <a href="http
Step2:  <a href="http
Step3:  <a href="http
Step4:  <a href="http
Step5:  <a href="http
Step6:  <a href="http
Step7:  <a href="http
Step8:  <a href="http
Step9:  <a href="http
Step10:  <a href="http
Step11:  <a href="http
Step12:  <a href="http
Step13:  <a href="http
Step14:  <a href="http
Step15:  <a href="http
Step16:  <a href="http
Step17:  <a href="http
Step18:  <a href="http
Step19:  <a href="http
Step20:  <a href="http
Step21:  <a href="http
Step22:  <a href="http
Step23:  <a href="http
Step24:  <a href="http
Step25:  <a href="http
Step26:  <a href="http
Step27:  <a href="http
Step28:  <a href="http
Step29:  <a href="http
Step30:  <a href="http
Step31:  <a href="http
Step32:  <a href="http
Step33:  <a href="http
Step34:  <a href="http
Step35:  <a href="http
Step36:  <a href="http
Step37:  <a href="http
Step38:  <a href="http
Step39:  <a href="http
Step40:  <a href="http
Step41:  <a href="http
Step42:  <a href="http
Step43:  <a href="http
Step44:  <a href="http
Step45:  <a href="http
Step46:  <a href="http
Step47:  <a href="http
Step48:  <a href="http
Step49:  <a href="http
Step50:  <a href="http
Step51:  <a href="http
Step52:  <a href="http
Step53:  <a href="http
Step54:  <a href="http
Step55:  <a href="http
Step56:  <a href="http
Step57:  <a href="http
Step58:  <a href="http
Step59:  <a href="http
Step60:  <a href="http
Step61:  <a href="http
Step62:  <a href="http
Step63:  <a href="http
Step64:  <a href="http
Step65:  <a href="http
<ASSISTANT_TASK:>
Python Code:
import IPython
print("pyspark version:" + str(sc.version))
print("Ipython version:" + str(IPython.__version__))
Explanation: <a>
<img align=left src="files/images/pyspark-page1.svg" width=500 height=250 />
</a>
DataFrame API
GitHub
related blog post
<a>
<img align=left src="files/images/pyspark-page2.svg" width=500 height=500>
</a>
Click on a picture to view pyspark docs
End of explanation
# map
x = sc.parallelize([1,2,3]) # sc = spark context, parallelize creates an RDD from the passed object
y = x.map(lambda x: (x,x**2))
print(x.collect())  # collect copies RDD elements to a list on the driver
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.map">
<img align=left src="files/images/pyspark-page3.svg" width=500 height=500 />
</a>
End of explanation
# flatMap
x = sc.parallelize([1,2,3])
y = x.flatMap(lambda x: (x, 100*x, x**2))
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.flatMap">
<img align=left src="files/images/pyspark-page4.svg" width=500 height=500 />
</a>
End of explanation
# mapPartitions
x = sc.parallelize([1,2,3], 2)
def f(iterator): yield sum(iterator)
y = x.mapPartitions(f)
print(x.glom().collect())  # glom() flattens elements on the same partition
print(y.glom().collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.mapPartitions">
<img align=left src="files/images/pyspark-page5.svg" width=500 height=500 />
</a>
End of explanation
# mapPartitionsWithIndex
x = sc.parallelize([1,2,3], 2)
def f(partitionIndex, iterator): yield (partitionIndex,sum(iterator))
y = x.mapPartitionsWithIndex(f)
print(x.glom().collect())  # glom() flattens elements on the same partition
print(y.glom().collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.mapPartitionsWithIndex">
<img align=left src="files/images/pyspark-page6.svg" width=500 height=500 />
</a>
End of explanation
# getNumPartitions
x = sc.parallelize([1,2,3], 2)
y = x.getNumPartitions()
print(x.glom().collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.getNumPartitions">
<img align=left src="files/images/pyspark-page7.svg" width=500 height=500 />
</a>
End of explanation
# filter
x = sc.parallelize([1,2,3])
y = x.filter(lambda x: x%2 == 1)  # filters out even elements
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.filter">
<img align=left src="files/images/pyspark-page8.svg" width=500 height=500 />
</a>
End of explanation
# distinct
x = sc.parallelize(['A','A','B'])
y = x.distinct()
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.distinct">
<img align=left src="files/images/pyspark-page9.svg" width=500 height=500 />
</a>
End of explanation
# sample
x = sc.parallelize(range(7))
ylist = [x.sample(withReplacement=False, fraction=0.5) for i in range(5)] # call 'sample' 5 times
print('x = ' + str(x.collect()))
for cnt,y in zip(range(len(ylist)), ylist):
    print('sample:' + str(cnt) + ' y = ' +  str(y.collect()))
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.sample">
<img align=left src="files/images/pyspark-page10.svg" width=500 height=500 />
</a>
End of explanation
# takeSample
x = sc.parallelize(range(7))
ylist = [x.takeSample(withReplacement=False, num=3) for i in range(5)]  # call 'sample' 5 times
print('x = ' + str(x.collect()))
for cnt,y in zip(range(len(ylist)), ylist):
    print('sample:' + str(cnt) + ' y = ' +  str(y))  # no collect on y
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.takeSample">
<img align=left src="files/images/pyspark-page11.svg" width=500 height=500 />
</a>
End of explanation
# union
x = sc.parallelize(['A','A','B'])
y = sc.parallelize(['D','C','A'])
z = x.union(y)
print(x.collect())
print(y.collect())
print(z.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.union">
<img align=left src="files/images/pyspark-page12.svg" width=500 height=500 />
</a>
End of explanation
# intersection
x = sc.parallelize(['A','A','B'])
y = sc.parallelize(['A','C','D'])
z = x.intersection(y)
print(x.collect())
print(y.collect())
print(z.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.intersection">
<img align=left src="files/images/pyspark-page13.svg" width=500 height=500 />
</a>
End of explanation
# sortByKey
x = sc.parallelize([('B',1),('A',2),('C',3)])
y = x.sortByKey()
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.sortByKey">
<img align=left src="files/images/pyspark-page14.svg" width=500 height=500 />
</a>
End of explanation
# sortBy
x = sc.parallelize(['Cat','Apple','Bat'])
def keyGen(val): return val[0]
y = x.sortBy(keyGen)
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.sortBy">
<img align=left src="files/images/pyspark-page15.svg" width=500 height=500 />
</a>
End of explanation
# glom
x = sc.parallelize(['C','B','A'], 2)
y = x.glom()
print(x.collect()) 
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.glom">
<img align=left src="files/images/pyspark-page16.svg" width=500 height=500 />
</a>
End of explanation
# cartesian
x = sc.parallelize(['A','B'])
y = sc.parallelize(['C','D'])
z = x.cartesian(y)
print(x.collect())
print(y.collect())
print(z.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.cartesian">
<img align=left src="files/images/pyspark-page17.svg" width=500 height=500 />
</a>
End of explanation
# groupBy
x = sc.parallelize([1,2,3])
y = x.groupBy(lambda x: 'A' if (x%2 == 1) else 'B' )
print(x.collect())
print([(j[0],[i for i in j[1]]) for j in y.collect()]) # y is nested, this iterates through it
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.groupBy">
<img align=left src="files/images/pyspark-page18.svg" width=500 height=500 />
<
End of explanation
# pipe
x = sc.parallelize(['A', 'Ba', 'C', 'AD'])
y = x.pipe('grep -i "A"') # calls out to grep, may fail under Windows 
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.pipe">
<img align=left src="files/images/pyspark-page19.svg" width=500 height=500 />
</a>
End of explanation
# foreach
from __future__ import print_function
x = sc.parallelize([1,2,3])
def f(el):
    '''side effect: append the current RDD elements to a file'''
    f1=open("./foreachExample.txt", 'a+') 
    print(el,file=f1)
open('./foreachExample.txt', 'w').close()  # first clear the file contents
y = x.foreach(f) # writes into foreachExample.txt
print(x.collect())
print(y) # foreach returns 'None'
# print the contents of foreachExample.txt
with open("./foreachExample.txt", "r") as foreachExample:
    print (foreachExample.read())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.foreach">
<img align=left src="files/images/pyspark-page20.svg" width=500 height=500 />
</a>
End of explanation
# foreachPartition
from __future__ import print_function
x = sc.parallelize([1,2,3],5)
def f(parition):
    '''side effect: append the current RDD partition contents to a file'''
    f1=open("./foreachPartitionExample.txt", 'a+') 
    print([el for el in parition],file=f1)
open('./foreachPartitionExample.txt', 'w').close()  # first clear the file contents
y = x.foreachPartition(f) # writes into foreachExample.txt
print(x.glom().collect())
print(y)  # foreach returns 'None'
# print the contents of foreachExample.txt
with open("./foreachPartitionExample.txt", "r") as foreachExample:
    print (foreachExample.read())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.foreachPartition">
<img align=left src="files/images/pyspark-page21.svg" width=500 height=500 />
</a>
End of explanation
# collect
x = sc.parallelize([1,2,3])
y = x.collect()
print(x)  # distributed
print(y)  # not distributed
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.collect">
<img align=left src="files/images/pyspark-page22.svg" width=500 height=500 />
</a>
End of explanation
# reduce
x = sc.parallelize([1,2,3])
y = x.reduce(lambda obj, accumulated: obj + accumulated)  # computes a cumulative sum
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.reduce">
<img align=left src="files/images/pyspark-page23.svg" width=500 height=500 />
</a>
End of explanation
# fold
x = sc.parallelize([1,2,3])
neutral_zero_value = 0  # 0 for sum, 1 for multiplication
y = x.fold(neutral_zero_value,lambda obj, accumulated: accumulated + obj) # computes cumulative sum
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.fold">
<img align=left src="files/images/pyspark-page24.svg" width=500 height=500 />
</a>
End of explanation
# aggregate
x = sc.parallelize([2,3,4])
neutral_zero_value = (0,1) # sum: x+0 = x, product: 1*x = x
seqOp = (lambda aggregated, el: (aggregated[0] + el, aggregated[1] * el)) 
combOp = (lambda aggregated, el: (aggregated[0] + el[0], aggregated[1] * el[1]))
y = x.aggregate(neutral_zero_value,seqOp,combOp)  # computes (cumulative sum, cumulative product)
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.aggregate">
<img align=left src="files/images/pyspark-page25.svg" width=500 height=500 />
</a>
End of explanation
# max
x = sc.parallelize([1,3,2])
y = x.max()
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.max">
<img align=left src="files/images/pyspark-page26.svg" width=500 height=500 />
</a>
End of explanation
# min
x = sc.parallelize([1,3,2])
y = x.min()
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.min">
<img align=left src="files/images/pyspark-page27.svg" width=500 height=500 />
</a>
End of explanation
# sum
x = sc.parallelize([1,3,2])
y = x.sum()
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.sum">
<img align=left src="files/images/pyspark-page28.svg" width=500 height=500 />
</a>
End of explanation
# count
x = sc.parallelize([1,3,2])
y = x.count()
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.count">
<img align=left src="files/images/pyspark-page29.svg" width=500 height=500 />
</a>
End of explanation
# histogram (example #1)
x = sc.parallelize([1,3,1,2,3])
y = x.histogram(buckets = 2)
print(x.collect())
print(y)
# histogram (example #2)
x = sc.parallelize([1,3,1,2,3])
y = x.histogram([0,0.5,1,1.5,2,2.5,3,3.5])
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.histogram">
<img align=left src="files/images/pyspark-page30.svg" width=500 height=500 />
</a>
End of explanation
# mean
x = sc.parallelize([1,3,2])
y = x.mean()
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.mean">
<img align=left src="files/images/pyspark-page31.svg" width=500 height=500 />
</a>
End of explanation
# variance
x = sc.parallelize([1,3,2])
y = x.variance()  # divides by N
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.variance">
<img align=left src="files/images/pyspark-page32.svg" width=500 height=500 />
</a>
End of explanation
# stdev
x = sc.parallelize([1,3,2])
y = x.stdev()  # divides by N
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.stdev">
<img align=left src="files/images/pyspark-page33.svg" width=500 height=500 />
</a>
End of explanation
# sampleStdev
x = sc.parallelize([1,3,2])
y = x.sampleStdev() # divides by N-1
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.sampleStdev">
<img align=left src="files/images/pyspark-page34.svg" width=500 height=500 />
</a>
End of explanation
# sampleVariance
x = sc.parallelize([1,3,2])
y = x.sampleVariance()  # divides by N-1
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.sampleVariance">
<img align=left src="files/images/pyspark-page35.svg" width=500 height=500 />
</a>
End of explanation
# countByValue
x = sc.parallelize([1,3,1,2,3])
y = x.countByValue()
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.countByValue">
<img align=left src="files/images/pyspark-page36.svg" width=500 height=500 />
</a>
End of explanation
# top
x = sc.parallelize([1,3,1,2,3])
y = x.top(num = 3)
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.top">
<img align=left src="files/images/pyspark-page37.svg" width=500 height=500 />
</a>
End of explanation
# takeOrdered
x = sc.parallelize([1,3,1,2,3])
y = x.takeOrdered(num = 3)
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.takeOrdered">
<img align=left src="files/images/pyspark-page38.svg" width=500 height=500 />
</a>
End of explanation
# take
x = sc.parallelize([1,3,1,2,3])
y = x.take(num = 3)
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.take">
<img align=left src="files/images/pyspark-page39.svg" width=500 height=500 />
</a>
End of explanation
# first
x = sc.parallelize([1,3,1,2,3])
y = x.first()
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.first">
<img align=left src="files/images/pyspark-page40.svg" width=500 height=500 />
</a>
End of explanation
# collectAsMap
x = sc.parallelize([('C',3),('A',1),('B',2)])
y = x.collectAsMap()
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.collectAsMap">
<img align=left src="files/images/pyspark-page41.svg" width=500 height=500 />
</a>
End of explanation
# keys
x = sc.parallelize([('C',3),('A',1),('B',2)])
y = x.keys()
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.keys">
<img align=left src="files/images/pyspark-page42.svg" width=500 height=500 />
</a>
End of explanation
# values
x = sc.parallelize([('C',3),('A',1),('B',2)])
y = x.values()
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.values">
<img align=left src="files/images/pyspark-page43.svg" width=500 height=500 />
</a>
End of explanation
# reduceByKey
x = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])
y = x.reduceByKey(lambda agg, obj: agg + obj)
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.reduceByKey">
<img align=left src="files/images/pyspark-page44.svg" width=500 height=500 />
</a>
End of explanation
# reduceByKeyLocally
x = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])
y = x.reduceByKeyLocally(lambda agg, obj: agg + obj)
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.reduceByKeyLocally">
<img align=left src="files/images/pyspark-page45.svg" width=500 height=500 />
</a>
End of explanation
# countByKey
x = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])
y = x.countByKey()
print(x.collect())
print(y)
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.countByKey">
<img align=left src="files/images/pyspark-page46.svg" width=500 height=500 />
</a>
End of explanation
# join
x = sc.parallelize([('C',4),('B',3),('A',2),('A',1)])
y = sc.parallelize([('A',8),('B',7),('A',6),('D',5)])
z = x.join(y)
print(x.collect())
print(y.collect())
print(z.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.join">
<img align=left src="files/images/pyspark-page47.svg" width=500 height=500 />
</a>
End of explanation
# leftOuterJoin
x = sc.parallelize([('C',4),('B',3),('A',2),('A',1)])
y = sc.parallelize([('A',8),('B',7),('A',6),('D',5)])
z = x.leftOuterJoin(y)
print(x.collect())
print(y.collect())
print(z.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.leftOuterJoin">
<img align=left src="files/images/pyspark-page48.svg" width=500 height=500 />
</a>
End of explanation
# rightOuterJoin
x = sc.parallelize([('C',4),('B',3),('A',2),('A',1)])
y = sc.parallelize([('A',8),('B',7),('A',6),('D',5)])
z = x.rightOuterJoin(y)
print(x.collect())
print(y.collect())
print(z.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.rightOuterJoin">
<img align=left src="files/images/pyspark-page49.svg" width=500 height=500 />
</a>
End of explanation
# partitionBy
x = sc.parallelize([(0,1),(1,2),(2,3)],2)
y = x.partitionBy(numPartitions = 3, partitionFunc = lambda x: x)  # only key is passed to paritionFunc
print(x.glom().collect())
print(y.glom().collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.partitionBy">
<img align=left src="files/images/pyspark-page50.svg" width=500 height=500 />
</a>
End of explanation
# combineByKey
x = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])
createCombiner = (lambda el: [(el,el**2)]) 
mergeVal = (lambda aggregated, el: aggregated + [(el,el**2)]) # append to aggregated
mergeComb = (lambda agg1,agg2: agg1 + agg2 )  # append agg1 with agg2
y = x.combineByKey(createCombiner,mergeVal,mergeComb)
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.combineByKey">
<img align=left src="files/images/pyspark-page51.svg" width=500 height=500 />
</a>
End of explanation
# aggregateByKey
x = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])
zeroValue = [] # empty list is 'zero value' for append operation
mergeVal = (lambda aggregated, el: aggregated + [(el,el**2)])
mergeComb = (lambda agg1,agg2: agg1 + agg2 )
y = x.aggregateByKey(zeroValue,mergeVal,mergeComb)
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.aggregateByKey">
<img align=left src="files/images/pyspark-page52.svg" width=500 height=500 />
</a>
End of explanation
# foldByKey
x = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])
zeroValue = 1 # one is 'zero value' for multiplication
y = x.foldByKey(zeroValue,lambda agg,x: agg*x )  # computes cumulative product within each key
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.foldByKey">
<img align=left src="files/images/pyspark-page53.svg" width=500 height=500 />
</a>
End of explanation
# groupByKey
x = sc.parallelize([('B',5),('B',4),('A',3),('A',2),('A',1)])
y = x.groupByKey()
print(x.collect())
print([(j[0],[i for i in j[1]]) for j in y.collect()])
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.groupByKey">
<img align=left src="files/images/pyspark-page54.svg" width=500 height=500 />
</a>
End of explanation
# flatMapValues
x = sc.parallelize([('A',(1,2,3)),('B',(4,5))])
y = x.flatMapValues(lambda x: [i**2 for i in x]) # function is applied to entire value, then result is flattened
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.flatMapValues">
<img align=left src="files/images/pyspark-page55.svg" width=500 height=500 />
</a>
End of explanation
# mapValues
x = sc.parallelize([('A',(1,2,3)),('B',(4,5))])
y = x.mapValues(lambda x: [i**2 for i in x]) # function is applied to entire value
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.mapValues">
<img align=left src="files/images/pyspark-page56.svg" width=500 height=500 />
</a>
End of explanation
# groupWith
x = sc.parallelize([('C',4),('B',(3,3)),('A',2),('A',(1,1))])
y = sc.parallelize([('B',(7,7)),('A',6),('D',(5,5))])
z = sc.parallelize([('D',9),('B',(8,8))])
a = x.groupWith(y,z)
print(x.collect())
print(y.collect())
print(z.collect())
print("Result:")
for key,val in list(a.collect()): 
    print(key, [list(i) for i in val])
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.groupWith">
<img align=left src="files/images/pyspark-page57.svg" width=500 height=500 />
</a>
End of explanation
# cogroup
x = sc.parallelize([('C',4),('B',(3,3)),('A',2),('A',(1,1))])
y = sc.parallelize([('A',8),('B',7),('A',6),('D',(5,5))])
z = x.cogroup(y)
print(x.collect())
print(y.collect())
for key,val in list(z.collect()):
    print(key, [list(i) for i in val])
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.cogroup">
<img align=left src="files/images/pyspark-page58.svg" width=500 height=500 />
</a>
End of explanation
# sampleByKey
x = sc.parallelize([('A',1),('B',2),('C',3),('B',4),('A',5)])
y = x.sampleByKey(withReplacement=False, fractions={'A':0.5, 'B':1, 'C':0.2})
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.sampleByKey">
<img align=left src="files/images/pyspark-page59.svg" width=500 height=500 />
</a>
End of explanation
# subtractByKey
x = sc.parallelize([('C',1),('B',2),('A',3),('A',4)])
y = sc.parallelize([('A',5),('D',6),('A',7),('D',8)])
z = x.subtractByKey(y)
print(x.collect())
print(y.collect())
print(z.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.subtractByKey">
<img align=left src="files/images/pyspark-page60.svg" width=500 height=500 />
</a>
End of explanation
# subtract
x = sc.parallelize([('C',4),('B',3),('A',2),('A',1)])
y = sc.parallelize([('C',8),('A',2),('D',1)])
z = x.subtract(y)
print(x.collect())
print(y.collect())
print(z.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.subtract">
<img align=left src="files/images/pyspark-page61.svg" width=500 height=500 />
</a>
End of explanation
# keyBy
x = sc.parallelize([1,2,3])
y = x.keyBy(lambda x: x**2)
print(x.collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.keyBy">
<img align=left src="files/images/pyspark-page62.svg" width=500 height=500 />
</a>
End of explanation
# repartition
x = sc.parallelize([1,2,3,4,5],2)
y = x.repartition(numPartitions=3)
print(x.glom().collect())
print(y.glom().collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.repartition">
<img align=left src="files/images/pyspark-page63.svg" width=500 height=500 />
</a>
End of explanation
# coalesce
x = sc.parallelize([1,2,3,4,5],2)
y = x.coalesce(numPartitions=1)
print(x.glom().collect())
print(y.glom().collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.coalesce">
<img align=left src="files/images/pyspark-page64.svg" width=500 height=500 />
</a>
End of explanation
# zip
x = sc.parallelize(['B','A','A'])
y = x.map(lambda x: ord(x))  # zip expects x and y to have same #partitions and #elements/partition
z = x.zip(y)
print(x.collect())
print(y.collect())
print(z.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.zip">
<img align=left src="files/images/pyspark-page65.svg" width=500 height=500 />
</a>
End of explanation
# zipWithIndex
x = sc.parallelize(['B','A','A'],2)
y = x.zipWithIndex()
print(x.glom().collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.zipWithIndex">
<img align=left src="files/images/pyspark-page66.svg" width=500 height=500 />
</a>
End of explanation
# zipWithUniqueId
x = sc.parallelize(['B','A','A'],2)
y = x.zipWithUniqueId()
print(x.glom().collect())
print(y.collect())
Explanation: <a href="http://spark.apache.org/docs/1.2.0/api/python/pyspark.html#pyspark.RDD.zipWithUniqueId">
<img align=left src="files/images/pyspark-page67.svg" width=500 height=500 />
</a>
End of explanation
<END_TASK>