{"id":8915,"date":"2026-09-21T05:41:44","date_gmt":"2026-09-21T05:41:44","guid":{"rendered":"https:\/\/prwatech.in\/blog\/?p=8915"},"modified":"2026-09-22T16:22:11","modified_gmt":"2026-09-22T16:22:11","slug":"working-with-dataproc","status":"publish","type":"post","link":"https:\/\/prwatech.in\/blog\/google-cloud-platform\/dataproc\/working-with-dataproc\/","title":{"rendered":"Working with Dataproc"},"content":{"rendered":"\r\n

Working With Dataproc: Four Spark Examples Over SSH<\/h1>\r\n

This runs four small Spark jobs directly on a Dataproc cluster’s master node over SSH, each a bit more involved than the last, counting movie ratings, totaling customer spending, finding the single most popular movie, then finding the top ten with their actual names attached. It complements the Submitting a Spark Job Through the Dataproc Console guide in this series, which shows the same kind of job submitted through the console interface instead of run directly on the cluster.<\/p>\r\n

This assumes a Dataproc cluster is already running. This walks through connecting to it, confirming Spark and Hive are available, then working through all four examples in order, since each introduces one new idea on top of the last.<\/p>\r\n

Complete Process of Working With Dataproc<\/h2>\r\n

Step 1: Connect to Your Cluster<\/h3>\r\n

Open the console, then<\/p>\r\n\r\n\r\n\r\n

Open Menu > Dataproc<\/a> > Clusters<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Click your cluster.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Click VM Instances.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Click SSH next to the master node.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Step 2: Confirm Spark and Hive Are Available<\/h3>\r\n

pyspark<\/p>\r\n

This opens the PySpark shell if it is installed. Exit with control + d.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Confirms Hive is available the same way. Exit with control d.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

python -V<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

spark-shell<\/p>\r\n

Opens the Spark shell. Exit with control d.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Example One: Counting Ratings<\/h2>\r\n

pwd
mkdir ratingscounter
cd ratingscounter<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

wget https:\/\/s3.amazonaws.com\/sankethadoop\/u.data
ls<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

nano u.data<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

The file’s contents are shown. Exit with control x.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

nano ratingscounter.py<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Paste in the following, which counts how many ratings fall into each score:<\/p>\r\n

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():
\u00a0\u00a0\u00a0\u00a0print(“%s %i” % (key, value))<\/p>\r\n

If you rename the directory or file, update the path inside the script to match. Save and exit with control x, then y, then enter.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Create Schema structure<\/p>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 hadoop fs -mkdir \/user\/<userid>\/sparkdata\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #To create directory named sparkdata<\/p>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 hadoop fs -put u.data sparkdata\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #To copy u.data file into sparkdata<\/p>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 hadoop fs -ls sparkdata\u00a0\u00a0 \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 \u00a0# to check the file is saved or not<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

spark-submit ratingscounter.py<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Example Two: Total Spend by Customer<\/h2>\r\n

cd
mkdir totalspendbycustomer
cd totalspendbycustomer<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

wget https:\/\/s3.amazonaws.com\/sankethadoop\/customer-orders.csv<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

ls
nano customer-orders.csv<\/p>\r\n

Confirm the file’s contents, then exit with control x.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Opens the file.<\/p>\r\n\r\n\r\n\r\n

To exit press ctrl +x<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 pwd\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #Displays the path<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 nano totalspendbycustomer.py\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #creates and opens file totalspendbycustomer.py<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Paste in the following, which totals how much each customer spent:<\/p>\r\n

from pyspark import SparkConf, SparkContext

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

def extractCustomerPricePairs(line):
\u00a0\u00a0\u00a0\u00a0fields = line.split(“,”)
\u00a0\u00a0\u00a0\u00a0return (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:
\u00a0\u00a0\u00a0\u00a0print(result)<\/p>\r\n

Save and exit the same way as before.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

hadoop fs -put customer-orders.csv sparkdata
hadoop fs -ls sparkdata
spark-submit totalspendbycustomer.py<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Each customer’s ID and total spend is displayed.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Example Three: The Single Most Popular Movie<\/h2>\r\n

cd
mkdir popularmovies
cd popularmovies<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 wget https:\/\/s3.amazonaws.com\/sankethadoop\/u.data \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #To copy file to disk<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 nano popularmovies.py\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #Open file popularmovies.py<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Paste in the following, which counts votes per movie and sorts by popularity:<\/p>\r\n

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:
\u00a0\u00a0\u00a0\u00a0print(result)<\/p>\r\n

Save and exit, then load the data and run it:<\/p>\r\n

hadoop fs -put u.data sparkdata
spark-submit popularmovies.py<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

It will show the most popular movie ID and most number of votes.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

Example Four: The Top Ten Popular Movies, With Names<\/h2>\r\n

cd
mkdir 10popularmovies
cd 10popularmovies<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

wget https:\/\/s3.amazonaws.com\/sankethadoop\/u.item
wget https:\/\/s3.amazonaws.com\/sankethadoop\/u.data<\/p>\r\n

u.item maps each movie ID to its actual title, which the earlier examples never used.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 nano u.item\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #To open the file content<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

To exit press ctrl+ x<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 nano 10popular.py\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 Create and open the file 10popular.py<\/p>\r\n\r\n\r\n\r\n

Paste the below code.<\/p>\r\n\r\n\r\n\r\n

from pyspark.sql import SparkSession<\/p>\r\n\r\n\r\n\r\n

from pyspark.sql import Row<\/p>\r\n\r\n\r\n\r\n

from pyspark.sql import functions<\/p>\r\n\r\n\r\n\r\n

def loadMovieNames():<\/p>\r\n\r\n\r\n\r\n

\u00a0\u00a0\u00a0\u00a0 movieNames = {}<\/p>\r\n\r\n\r\n\r\n

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

\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 for line in f:<\/p>\r\n\r\n\r\n\r\n

\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 fields = line.split(‘|’)<\/p>\r\n\r\n\r\n\r\n

\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 movieNames[int(fields[0])] = fields [1]<\/p>\r\n\r\n\r\n\r\n

\u00a0\u00a0\u00a0\u00a0 return movieNames<\/p>\r\n\r\n\r\n\r\n

spark = SparkSession.builder.appName(“PopularMovies”).getOrCreate()<\/p>\r\n\r\n\r\n\r\n

nameDict = loadMovieNames()<\/p>\r\n\r\n\r\n\r\n

lines = spark.sparkContext.textFile(“sparkdata\/u.data”)<\/p>\r\n\r\n\r\n\r\n

movies = lines.map(lambda x: Row(movieID =int(x.split()[1])))<\/p>\r\n\r\n\r\n\r\n

movieDataset = spark.createDataFrame(movies)<\/p>\r\n\r\n\r\n\r\n

topMovieIDs = movieDataset.groupBy(“movieID”).count().orderBy(“count”,ascending = False).cache()<\/p>\r\n\r\n\r\n\r\n

topMovieIDs.show()<\/p>\r\n\r\n\r\n\r\n

top10 = topMovieIDs.take(10)<\/p>\r\n\r\n\r\n\r\n

print(“\\n”)<\/p>\r\n\r\n\r\n\r\n

for result in top10:<\/p>\r\n\r\n\r\n\r\n

\u00a0\u00a0\u00a0\u00a0 print(“%s: %d” % (nameDict[result[0]], result[1]))<\/p>\r\n\r\n\r\n\r\n

spark.stop()<\/p>\r\n\r\n\r\n\r\n

Change the highlighted area as your directory<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 hadoop fs -put u.data sparkdata\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #To copy u.data file into sparkdata<\/p>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 hadoop fs -ls sparkdata\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #To display the content<\/p>\r\n\r\n\r\n\r\n

$\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 spark-submit 10popular.py\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 #To execute the 10popular.py file<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n\r\n\r\n\r\n

It will display the most popular 10 movies.<\/p>\r\n\r\n\r\n\r\n

\"\"<\/figure>\r\n

How These Four Examples Build on Each Other<\/h2>\r\n

Each example adds exactly one new idea rather than starting from scratch. The first counts values directly. The second maps each line into a key and value pair before combining them. The third flips a result to sort by count instead of by key. The fourth brings in a second file entirely, joining an ID based result against a lookup table to show real names instead of raw numbers, using the newer DataFrame API rather than plain RDDs. Working through them in order is worth more than jumping straight to the last one.<\/p>\r\n

Common Mistakes to Avoid<\/h2>\r\n