Prerequisites

GCP account

Open Cloud Console.

Open Menu > Dataproc > Clusters

Click the Cluster.

Click on VM Instances

Click on SSH of master node

Check whether the components is already installed.

$          pyspark                     #opens pyspark

To exit press ctrl +d

$          hive                #To check hive is available or not

To exit press ctrl +d

$          python –V                #to check python version

$          spark-shell                # opens spark shell

To exit press ctrl +d

$          pwd                                        #To get path

$          mkdir ratingscounter        #making directory named ratingscounter

$          cd ratingscounter               #Change the directory into ratingscounter

$          wget https://s3.amazonaws.com/sankethadoop/u.data              #To get the data for dataproc

$          ls                                 #Display the contents in the directory

$          nano u.data                         #open the u.data file.

It will display the content in u.data

To exit press ctrl + x

$          nano ratingscounter.py               #Creates and opens file ratingscounter.py

Paste the below code into ratingscounter.py file

from pyspark import SparkConf, SparkContext

import collections

conf = SparkConf().setMaster(“local”).setAppName(“Ratings”)

sc = SparkContext(conf = conf)

lines = sc.textFile(“sparkdata/u.data”)

ratings = lines.map(lambda x: x.split( )[2])

result = ratings.countByValue()

sortedResults = collections.OrderedDict(sorted(result.items()))

for key, value in sortedResults.items():

    print(“%s %i” % (key, value))

This code is to count the films in each ratings.

NB : if you are changing the name of directory or file, you may have to change it in the file lolcation also.

To exit press ‘ctrl + x’ then press ‘y’ to confirm then ‘Enter’

Create Schema structure

$          hadoop fs -mkdir /user/<userid>/sparkdata             #To create directory named sparkdata

$          hadoop fs -put u.data sparkdata                                   #To copy u.data file into sparkdata

$          hadoop fs -ls sparkdata                                                     # to check the file is saved or not

$          spark-submit ratingscounter.py                                    #execute the ratingscounter.py file

It will display the result.

$          cd                                                                    #change directory

$          mkdir totalspendbycustomer                 #make directory

$          cd totalspendbycustomer                        #change directory to totalspendbycustomer

$          wget https://s3.amazonaws.com/sankethadoop/customer-orders.csv           #To copy file to disk

$          ls                                                                     #list the contents in the directory

$          nano customer-orders.csv                      #Open the file customer-orders.csv

Opens the file.

To exit press ctrl +x

$          pwd                            #Displays the path

$          nano totalspendbycustomer.py                        #creates and opens file totalspendbycustomer.py

paste the below code

from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster(“local”).setAppName(“SpendByCustomer”)

sc = SparkContext(conf = conf)

def extractCustomerPricePairs(line):

    fields = line.split(‘,’)

    return (int(fields[0]), float(fields[2]))

input = sc.textFile(“sparkdata/customer-orders.csv”)

mappedInput = input.map(extractCustomerPricePairs)

totalByCustomer = mappedInput.reduceByKey(lambda x, y: x + y)

results = totalByCustomer.collect();

for result in results:

    print(result)

This code is to get the amount spent by the customers for movie

NB : if you are changing the name of directory or file, you may have to change it in the file lolcation also.

To exit press ‘ctrl + x’ then press ‘y’ to confirm then ‘Enter’

$          hadoop fs -put customer-orders.csv sparkdata                    #moves file customer-orders.csv into sparkdata

$          hadoop fs -ls sparkdata                                                    # to check the file is saved or not

$          spark-submit totalspendbycustomer.py                     #Execute the file totalspendbycustomer.py

It will display the customer ID and total amount spend by customer

To find popular movies

$          cd                                                        #To Change Directory

$          mkdir popularmovies                   #Make directory named popularmovies

$          cd popularmovies                          #To change directory into popularmovies

$          wget https://s3.amazonaws.com/sankethadoop/u.data                           #To copy file to disk

$          nano popularmovies.py                                       #Open file popularmovies.py

In the python file paste the below code.

from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster(“local”).setAppName(“PopularMovies”)

sc = SparkContext(conf = conf)

lines = sc.textFile(“sparkdata/u.data”)

movies = lines.map(lambda x: (int(x.split()[1]), 1))

movieCounts = movies.reduceByKey(lambda x, y: x + y)

flipped = movieCounts.map( lambda xy: (xy[1],xy[0]) )

sortedMovies = flipped.sortByKey()

results = sortedMovies.collect()

for result in results:

     print(result)

To save and exit, Press ‘Ctrl + x’ then ‘y’ then ‘Enter’

$          hadoop fs -put u.data sparkdata                                   #To copy u.data file into sparkdata

$          spark-submit popularmovies.py                                                # To execute popularmovies.py

It will show the most popular movie ID and most number of votes.

To find most 10 popular movies

$          cd                                                                    #To Change Directory

$          mkdir 10popularmovies                          #Make directory named popularmovies

$          cd 10popularmovies                                 #To change directory into popularmovies

$          wget https://s3.amazonaws.com/sankethadoop/u.item

$          wget https://s3.amazonaws.com/sankethadoop/u.data

It will copy the file into disk.

$          nano u.item                                     #To open the file content

To exit press ctrl+ x

$          nano 10popular.py                                   Create and open the file 10popular.py

Paste the below code.

from pyspark.sql import SparkSession

from pyspark.sql import Row

from pyspark.sql import functions

def loadMovieNames():

     movieNames = {}

     with open(“/home/<userid>/10popularmovies/u.item“, encoding=”ISO-8859-1”) as f:

          for line in f:

              fields = line.split(‘|’)

              movieNames[int(fields[0])] = fields [1]

     return movieNames

spark = SparkSession.builder.appName(“PopularMovies”).getOrCreate()

nameDict = loadMovieNames()

lines = spark.sparkContext.textFile(“sparkdata/u.data”)

movies = lines.map(lambda x: Row(movieID =int(x.split()[1])))

movieDataset = spark.createDataFrame(movies)

topMovieIDs = movieDataset.groupBy(“movieID”).count().orderBy(“count”,ascending = False).cache()

topMovieIDs.show()

top10 = topMovieIDs.take(10)

print(“\n”)

for result in top10:

     print(“%s: %d” % (nameDict[result[0]], result[1]))

spark.stop()

Change the highlighted area as your directory

$          hadoop fs -put u.data sparkdata                                   #To copy u.data file into sparkdata

$          hadoop fs -ls sparkdata                                                    #To display the content

$          spark-submit 10popular.py                                            #To execute the 10popular.py file

It will display the most popular 10 movies.