%SPARK_HOME%\python;%SPARK_HOME%\python\lib\py4j-0.10.9.7-src.zip;%PYTHONPATH%
C:\>pyspark
Python 3.11.7 | packaged by Anaconda, Inc. | (main, Dec 15 2023, 18:05:47) [MSC v.1916 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
24/04/24 14:18:23 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/__ / .__/\_,_/_/ /_/\_\ version 3.5.1
/_/
Using Python version 3.11.7 (main, Dec 15 2023 18:05:47)
Spark context Web UI available at http://10.0.0.202:4041
Spark context available as 'sc' (master = local[*], app id = local-1713993503365).
SparkSession available as 'spark'.
>>> rdd1 = sc.textFile("C:/Users/User/a2.txt")
>>> rdd1.collect()
...
['alex,Cupertino,5', 'jane,Cupertino,8', 'jeff,Cupertino,6', 'jeff,Cupertino,8', 'alex,Sunnyvale,10', 'jane,Sunnyvale,8', 'jeff,Sunnyvale,3', 'alex,Stanford,8', 'jane,Stanford,8', 'jeff,Stanford,10', 'jeff,Stanford,10', 'bob,Sunnyvale,10', 'rafa,Sunnyvale,10', 'jane,Sunnyvale,4', 'jane,Sunnyvale,10', 'david,Ames,8', 'alex,Ames,10', 'jane,Ames,10', 'jeff,Ames,10', 'jeff,Saratoga,8', 'alex,Saratoga,1', 'jane,Saratoga,7', 'jeff,Saratoga,4', 'alex,Stanford,8', 'jane,Stanford,8', 'jeff,Stanford,10', 'jeff,Stanford,10', 'bob,Sunnyvale,10', 'rafa,Oakland,34', 'jane,Oakland,4', 'jane,Oakland,1', 'jane,Fremont,6', 'jane,Fremont,3', 'jane,Fremont,1']
>>> def tokenize(record):
... tokens = record.split(',')
... city = tokens[1]
... rating = float(tokens[2])
... return (city, rating)
...
>>> tokenize('Cupertino,Jake,3')
('Jake', 3.0)
>>> pairs = rdd1.map(tokenize)
>>> pairs.collect()
[('Cupertino', 5.0), ('Cupertino', 8.0), ('Cupertino', 6.0), ('Cupertino', 8.0), ('Sunnyvale', 10.0), ('Sunnyvale', 8.0), ('Sunnyvale', 3.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('Stanford', 10.0), ('Sunnyvale', 10.0), ('Sunnyvale', 10.0), ('Sunnyvale', 4.0), ('Sunnyvale', 10.0), ('Ames', 8.0), ('Ames', 10.0), ('Ames', 10.0), ('Ames', 10.0), ('Saratoga', 8.0), ('Saratoga', 1.0), ('Saratoga', 7.0), ('Saratoga', 4.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('Stanford', 10.0), ('Sunnyvale', 10.0), ('Oakland', 34.0), ('Oakland', 4.0), ('Oakland', 1.0), ('Fremont', 6.0), ('Fremont', 3.0), ('Fremont', 1.0)]
>>> filtered = pairs.filter(lambda x: x[1] >=2 and x[1] < 11)
>>> filtered.collect()
[Stage 2:> (0 + 2) [('Cupertino', 5.0), ('Cupertino', 8.0), ('Cupertino', 6.0), ('Cupertino', 8.0), ('Sunnyvale', 10.0), ('Sunnyvale', 8.0), ('Sunnyvale', 3.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('Stanford', 10.0), ('Sunnyvale', 10.0), ('Sunnyvale', 10.0), ('Sunnyvale', 4.0), ('Sunnyvale', 10.0), ('Ames', 8.0), ('Ames', 10.0), ('Ames', 10.0), ('Ames', 10.0), ('Saratoga', 8.0), ('Saratoga', 7.0), ('Saratoga', 4.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('Stanford', 10.0), ('Sunnyvale', 10.0), ('Oakland', 4.0), ('Fremont', 6.0), ('Fremont', 3.0)]
>>> def flatten(keyvalue):
... city = keyvalue[0]
... rating = keyvalue[1]
... if rating == 10:
... return [(city, rating), ("unique-10", city)]
... else:
... return [(city, rating)]
...
>>> flattened = filtered.flatMap(flatten)
>>> flattened.collect()
[Stage 3:> (0 + 2)
[('Cupertino', 5.0), ('Cupertino', 8.0), ('Cupertino', 6.0), ('Cupertino', 8.0), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Sunnyvale', 8.0), ('Sunnyvale', 3.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('unique-10', 'Stanford'), ('Stanford', 10.0), ('unique-10', 'Stanford'), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Sunnyvale', 4.0), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Ames', 8.0), ('Ames', 10.0), ('unique-10', 'Ames'), ('Ames', 10.0), ('unique-10', 'Ames'), ('Ames', 10.0), ('unique-10', 'Ames'), ('Saratoga', 8.0), ('Saratoga', 7.0), ('Saratoga', 4.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('unique-10', 'Stanford'), ('Stanford', 10.0), ('unique-10', 'Stanford'), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Oakland', 4.0), ('Fremont', 6.0), ('Fremont', 3.0)]
>>> grouped = flattened.groupByKey()
>>> grouped.mapValues(lambda values: list(values)).collect()
[Stage 4:> (0 + 2) [Stage 5:> (0 + 2)
[('Sunnyvale', [10.0, 8.0, 3.0, 10.0, 10.0, 4.0, 10.0, 10.0]), ('Saratoga', [8.0, 7.0, 4.0]), ('Fremont', [6.0, 3.0]), ('Cupertino', [5.0, 8.0, 6.0, 8.0]), ('unique-10', ['Sunnyvale', 'Stanford', 'Stanford', 'Sunnyvale', 'Sunnyvale', 'Sunnyvale', 'Ames', 'Ames', 'Ames', 'Stanford', 'Stanford', 'Sunnyvale']), ('Stanford', [8.0, 8.0, 10.0, 10.0, 8.0, 8.0, 10.0, 10.0]), ('Ames', [8.0, 10.0, 10.0, 10.0]), ('Oakland', [4.0])]
>>> grouped = flattened.groupByKey()
>>> grouped.mapValues(lambda values: list(values)).collect()
[Stage 4:> (0 + 2) [Stage 5:> (0 + 2)
[('Sunnyvale', [10.0, 8.0, 3.0, 10.0, 10.0, 4.0, 10.0, 10.0]), ('Saratoga', [8.0, 7.0, 4.0]), ('Fremont', [6.0, 3.0]), ('Cupertino', [5.0, 8.0, 6.0, 8.0]), ('unique-10', ['Sunnyvale', 'Stanford', 'Stanford', 'Sunnyvale', 'Sunnyvale', 'Sunnyvale', 'Ames', 'Ames', 'Ames', 'Stanford', 'Stanford', 'Sunnyvale']), ('Stanford', [8.0, 8.0, 10.0, 10.0, 8.0, 8.0, 10.0, 10.0]), ('Ames', [8.0, 10.0, 10.0, 10.0]), ('Oakland', [4.0])]
>>> # create a reduction function
>>> def reduction(keyvalue):
... key = keyvale[0]
... values = keyvalue[1]
... if key == "unique-10":
... return (key, set(values))
... else:
... avg = sum(values)/len(values)
... if (avg >= 4):
... return (key, avg)
...
>>> # apply reduction function
>>> result = grouped.map(reduction)
>>> result.collect()
C:\>pyspark
Python 3.11.7 | packaged by Anaconda, Inc. | (main, Dec 15 2023, 18:05:47) [MSC v.1916 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/__ / .__/\_,_/_/ /_/\_\ version 3.5.1
/_/
Using Python version 3.11.7 (main, Dec 15 2023 18:05:47)
Spark context Web UI available at http://10.0.0.202:4040
Spark context available as 'sc' (master = local[*], app id = local-1714419282255).
SparkSession available as 'spark'.
>>> rdd1 = sc.textFile("C:/Users/User/a2.txt")
>>> rdd1.collect()
[Stage 0:> (0 + 2) ['alex,Cupertino,5', 'jane,Cupertino,8', 'jeff,Cupertino,6', 'jeff,Cupertino,8', 'alex,Sunnyvale,10', 'jane,Sunnyvale,8', 'jeff,Sunnyvale,3', 'alex,Stanford,8', 'jane,Stanford,8', 'jeff,Stanford,10', 'jeff,Stanford,10', 'bob,Sunnyvale,10', 'rafa,Sunnyvale,10', 'jane,Sunnyvale,4', 'jane,Sunnyvale,10', 'david,Ames,8', 'alex,Ames,10', 'jane,Ames,10', 'jeff,Ames,10', 'jeff,Saratoga,8', 'alex,Saratoga,1', 'jane,Saratoga,7', 'jeff,Saratoga,4', 'alex,Stanford,8', 'jane,Stanford,8', 'jeff,Stanford,10', 'jeff,Stanford,10', 'bob,Sunnyvale,10', 'rafa,Oakland,34', 'jane,Oakland,4', 'jane,Oakland,1', 'jane,Fremont,6', 'jane,Fremont,3', 'jane,Fremont,1']
>>> def tokenize(record):
... tokens = record.split(",")
... city = tokens[1]
... rating = float(tokens[2])
... return (city, rating)
>>># test function
>>> tokenize('Jake,Cupertino,3')
('Cupertino', 3.0)
>>> pairs = rdd1.map(tokenize)
>>> pairs.collect()
[Stage 1:> (0 + 2)
[('Cupertino', 5.0), ('Cupertino', 8.0), ('Cupertino', 6.0), ('Cupertino', 8.0), ('Sunnyvale', 10.0), ('Sunnyvale', 8.0), ('Sunnyvale', 3.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('Stanford', 10.0), ('Sunnyvale', 10.0), ('Sunnyvale', 10.0), ('Sunnyvale', 4.0), ('Sunnyvale', 10.0), ('Ames', 8.0), ('Ames', 10.0), ('Ames', 10.0), ('Ames', 10.0), ('Saratoga', 8.0), ('Saratoga', 1.0), ('Saratoga', 7.0), ('Saratoga', 4.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('Stanford', 10.0), ('Sunnyvale', 10.0), ('Oakland', 34.0), ('Oakland', 4.0), ('Oakland', 1.0), ('Fremont', 6.0), ('Fremont', 3.0), ('Fremont', 1.0)]
# filter out based on rating
>>> filtered = pairs.filter(lambda x: x[1] >= 2 and x[1] < 11)
>>> filtered.collect()
[Stage 2:> (0 + 2) [Stage 2:=============================> (1 + 1)
[('Cupertino', 5.0), ('Cupertino', 8.0), ('Cupertino', 6.0), ('Cupertino', 8.0), ('Sunnyvale', 10.0), ('Sunnyvale', 8.0), ('Sunnyvale', 3.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('Stanford', 10.0), ('Sunnyvale', 10.0), ('Sunnyvale', 10.0), ('Sunnyvale', 4.0), ('Sunnyvale', 10.0), ('Ames', 8.0), ('Ames', 10.0), ('Ames', 10.0), ('Ames', 10.0), ('Saratoga', 8.0), ('Saratoga', 7.0), ('Saratoga', 4.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('Stanford', 10.0), ('Sunnyvale', 10.0), ('Oakland', 4.0), ('Fremont', 6.0), ('Fremont', 3.0)]
# flatMap transformation: 1-to-many
>>> def flatten(key_value):
... city = key_value[0]
... rating = key_value[1]
... if rating == 10:
... return [(city, rating), ("unique-10", city)]
... else:
... return [(city, rating)]
...
>>> flattened = filtered.flatMap(flatten)
>>> flattened.collect()
[Stage 3:> (0 + 2) [Stage 3:=============================> (1 + 1)
[('Cupertino', 5.0), ('Cupertino', 8.0), ('Cupertino', 6.0), ('Cupertino', 8.0), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Sunnyvale', 8.0), ('Sunnyvale', 3.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('unique-10', 'Stanford'), ('Stanford', 10.0), ('unique-10', 'Stanford'), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Sunnyvale', 4.0), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Ames', 8.0), ('Ames', 10.0), ('unique-10', 'Ames'), ('Ames', 10.0), ('unique-10', 'Ames'), ('Ames', 10.0), ('unique-10', 'Ames'), ('Saratoga', 8.0), ('Saratoga', 7.0), ('Saratoga', 4.0), ('Stanford', 8.0), ('Stanford', 8.0), ('Stanford', 10.0), ('unique-10', 'Stanford'), ('Stanford', 10.0), ('unique-10', 'Stanford'), ('Sunnyvale', 10.0), ('unique-10', 'Sunnyvale'), ('Oakland', 4.0), ('Fremont', 6.0), ('Fremont', 3.0)]
# groupByKey transformation -> SQL group by
>>> grouped = flattened.groupByKey()
>>> grouped.mapValues(lambda values: list(values)).collect()
[Stage 4:> (0 + 2) [Stage 4:=============================> (1 + 1) [Stage 5:> (0 + 2) [Stage 5:=============================> (1 + 1)
[('Sunnyvale', [10.0, 8.0, 3.0, 10.0, 10.0, 4.0, 10.0, 10.0]), ('Saratoga', [8.0, 7.0, 4.0]), ('Fremont', [6.0, 3.0]), ('Cupertino', [5.0, 8.0, 6.0, 8.0]), ('unique-10', ['Sunnyvale', 'Stanford', 'Stanford', 'Sunnyvale', 'Sunnyvale', 'Sunnyvale', 'Ames', 'Ames', 'Ames', 'Stanford', 'Stanford', 'Sunnyvale']), ('Stanford', [8.0, 8.0, 10.0, 10.0, 8.0, 8.0, 10.0, 10.0]), ('Ames', [8.0, 10.0, 10.0, 10.0]), ('Oakland', [4.0])]
# reduction function
>>> def reduction(key_value):
... key = key_value[0]
... values = key_value[1]
... if key == 'unique-10':
... return (key, set(values))
... else:
... avg = sum(values)/len(values)
... if (avg >= 4):
... return (key, avg)
...
>>> result = grouped.map(reduction)
>>> result.collect()
[Stage 7:> (0 + 2)
[('Sunnyvale', 8.125), ('Saratoga', 6.333333333333333), ('Fremont', 4.5), ('Cupertino', 6.75), ('unique-10', {'Sunnyvale', 'Ames', 'Stanford'}), ('Stanford', 9.0), ('Ames', 9.5), ('Oakland', 4.0)]