#!/usr/bin/env python
# coding: utf-8

# <a>
# <img align=left src="files/images/pyspark-page1.svg" width=500 height=250 />
# </a>

# * [DataFrame API](http://nbviewer.ipython.org/github/jkthompson/pyspark-pictures/blob/master/pyspark-pictures-dataframes.ipynb)
# * [GitHub](https://github.com/jkthompson/pyspark-pictures)
# * [related blog post](http://data-frack.blogspot.com/2015/01/visual-mnemonics-for-pyspark-api.html)

# <a>
# <img align=left src="files/images/pyspark-page2.svg" width=500 height=500>
# </a>

# # Click on a picture to view pyspark docs

# In[1]:


import IPython
print("pyspark version:" + str(sc.version))
print("Ipython version:" + str(IPython.__version__))


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

# In[2]:


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


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

# In[3]:


# flatMap
x = sc.parallelize([1,2,3])
y = x.flatMap(lambda x: (x, 100*x, x**2))
print(x.collect())
print(y.collect())


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

# In[4]:


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


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

# In[5]:


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


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

# In[6]:


# getNumPartitions
x = sc.parallelize([1,2,3], 2)
y = x.getNumPartitions()
print(x.glom().collect())
print(y)


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

# In[7]:


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


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

# In[8]:


# distinct
x = sc.parallelize(['A','A','B'])
y = x.distinct()
print(x.collect())
print(y.collect())


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

# In[9]:


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


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

# In[10]:


# 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


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

# In[11]:


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


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

# In[12]:


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


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

# In[13]:


# sortByKey
x = sc.parallelize([('B',1),('A',2),('C',3)])
y = x.sortByKey()
print(x.collect())
print(y.collect())


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

# In[14]:


# sortBy
x = sc.parallelize(['Cat','Apple','Bat'])
def keyGen(val): return val[0]
y = x.sortBy(keyGen)
print(y.collect())


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

# In[15]:


# glom
x = sc.parallelize(['C','B','A'], 2)
y = x.glom()
print(x.collect()) 
print(y.collect())


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

# In[16]:


# cartesian
x = sc.parallelize(['A','B'])
y = sc.parallelize(['C','D'])
z = x.cartesian(y)
print(x.collect())
print(y.collect())
print(z.collect())


# <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 />
# <

# In[17]:


# 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


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

# In[18]:


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


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

# In[19]:


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


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

# In[20]:


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


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

# In[21]:


# collect
x = sc.parallelize([1,2,3])
y = x.collect()
print(x)  # distributed
print(y)  # not distributed


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

# In[22]:


# reduce
x = sc.parallelize([1,2,3])
y = x.reduce(lambda obj, accumulated: obj + accumulated)  # computes a cumulative sum
print(x.collect())
print(y)


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

# In[23]:


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


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

# In[24]:


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


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

# In[25]:


# max
x = sc.parallelize([1,3,2])
y = x.max()
print(x.collect())
print(y)


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

# In[26]:


# min
x = sc.parallelize([1,3,2])
y = x.min()
print(x.collect())
print(y)


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

# In[27]:


# sum
x = sc.parallelize([1,3,2])
y = x.sum()
print(x.collect())
print(y)


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

# In[28]:


# count
x = sc.parallelize([1,3,2])
y = x.count()
print(x.collect())
print(y)


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

# In[29]:


# histogram (example #1)
x = sc.parallelize([1,3,1,2,3])
y = x.histogram(buckets = 2)
print(x.collect())
print(y)


# In[30]:


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


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

# In[31]:


# mean
x = sc.parallelize([1,3,2])
y = x.mean()
print(x.collect())
print(y)


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

# In[32]:


# variance
x = sc.parallelize([1,3,2])
y = x.variance()  # divides by N
print(x.collect())
print(y)


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

# In[33]:


# stdev
x = sc.parallelize([1,3,2])
y = x.stdev()  # divides by N
print(x.collect())
print(y)


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

# In[34]:


# sampleStdev
x = sc.parallelize([1,3,2])
y = x.sampleStdev() # divides by N-1
print(x.collect())
print(y)


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

# In[35]:


# sampleVariance
x = sc.parallelize([1,3,2])
y = x.sampleVariance()  # divides by N-1
print(x.collect())
print(y)


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

# In[36]:


# countByValue
x = sc.parallelize([1,3,1,2,3])
y = x.countByValue()
print(x.collect())
print(y)


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

# In[37]:


# top
x = sc.parallelize([1,3,1,2,3])
y = x.top(num = 3)
print(x.collect())
print(y)


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

# In[38]:


# takeOrdered
x = sc.parallelize([1,3,1,2,3])
y = x.takeOrdered(num = 3)
print(x.collect())
print(y)


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

# In[39]:


# take
x = sc.parallelize([1,3,1,2,3])
y = x.take(num = 3)
print(x.collect())
print(y)


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

# In[40]:


# first
x = sc.parallelize([1,3,1,2,3])
y = x.first()
print(x.collect())
print(y)


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

# In[41]:


# collectAsMap
x = sc.parallelize([('C',3),('A',1),('B',2)])
y = x.collectAsMap()
print(x.collect())
print(y)


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

# In[42]:


# keys
x = sc.parallelize([('C',3),('A',1),('B',2)])
y = x.keys()
print(x.collect())
print(y.collect())


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

# In[43]:


# values
x = sc.parallelize([('C',3),('A',1),('B',2)])
y = x.values()
print(x.collect())
print(y.collect())


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

# In[44]:


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


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

# In[45]:


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


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

# In[46]:


# countByKey
x = sc.parallelize([('B',1),('B',2),('A',3),('A',4),('A',5)])
y = x.countByKey()
print(x.collect())
print(y)


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

# In[47]:


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


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

# In[48]:


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


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

# In[49]:


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


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

# In[50]:


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


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

# In[51]:


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


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

# In[52]:


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


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

# In[53]:


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


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

# In[54]:


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


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

# In[55]:


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


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

# In[56]:


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


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

# In[57]:


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


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

# In[58]:


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


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

# In[59]:


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


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

# In[60]:


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


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

# In[61]:


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


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

# In[62]:


# keyBy
x = sc.parallelize([1,2,3])
y = x.keyBy(lambda x: x**2)
print(x.collect())
print(y.collect())


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

# In[63]:


# repartition
x = sc.parallelize([1,2,3,4,5],2)
y = x.repartition(numPartitions=3)
print(x.glom().collect())
print(y.glom().collect())


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

# In[64]:


# coalesce
x = sc.parallelize([1,2,3,4,5],2)
y = x.coalesce(numPartitions=1)
print(x.glom().collect())
print(y.glom().collect())


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

# In[65]:


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


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

# In[66]:


# zipWithIndex
x = sc.parallelize(['B','A','A'],2)
y = x.zipWithIndex()
print(x.glom().collect())
print(y.collect())


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

# In[67]:


# zipWithUniqueId
x = sc.parallelize(['B','A','A'],2)
y = x.zipWithUniqueId()
print(x.glom().collect())
print(y.collect())