<jupyter_start><jupyter_text>Why Spark? Because ... Pyspark==========![Relatedimage](https://miro.medium.com/proxy/1*sQGVLk43kXJTEw1mtJRoDw.png)Hadoop was the first open source system that introduced us to theMapReduce paradigm of programming and Spark is the system that made itfaster, much much faster(100x).There used to be a lot of data movement in Hadoop as it used to writeintermediate results to the file system.This affected the speed at which you could do analysis.Spark provided us with an in-memory model, so Spark doesn’t write toomuch to the disk while working.Simply, Spark is faster than Hadoop and a lot of people use Spark now.***So without further ado let us get started.***Load Some Data==============The next step is to upload some data we will use to learn Spark. We will end up using multiple datasets by the end of this but let us start with something very simple.Let us add the file `shakespeare.txt` You can see that the file is loaded to `shakespeare/shakespeare.txt` location.<jupyter_code># To download the data you would use the following commands:
#!wget -P shakespeare https://ocw.mit.edu/ans7870/6/6.006/s08/lecturenotes/files/t8.shakespeare.txt
#!mv shakespeare/t8.shakespeare.txt shakespeare/shakespeare.txt
!ls -l shakespeare<jupyter_output>total 5332
-rw-r--r-- 1 root root 5458199 Apr 23  2020 shakespeare.txt<jupyter_text>Our First Spark Program=======================I like to learn by examples so let’s get done with the “Hello World” ofDistributed computing: ***The WordCount Program.***First, we need to create a `SparkSession`:<jupyter_code>from pyspark.sql import SparkSession

spark = SparkSession\
        .builder\
        .appName("shakespeare")\
        .master("spark://spark-master:7077")\
        .config("hive.metastore.uris", "thrift://hive-metastore:9083")\
        .config("spark.sql.warehouse.dir", "hdfs://namenode:8020/user/hive/warehouse")\
        .config("spark.executor.memory", "1g")\
        .config("spark.jars.packages", "org.apache.spark:spark-avro_2.12:3.2.0")\
        .enableHiveSupport()\
        .getOrCreate()

sc = spark.sparkContext

sc.setLogLevel("ERROR")

from pyspark.sql.functions import to_json,col
from pyspark.sql.types import *
from os.path import abspath
# Distribute the data - Create a RDD 
lines = sc.textFile("shakespeare/shakespeare.txt")
x = 'This is the 100th Etext file presented by Project Gutenberg, and'
x.split(' ')<jupyter_output><empty_output><jupyter_text>Now we can write our program:<jupyter_code># Distribute the data - Create a RDD 
lines = sc.textFile("shakespeare/shakespeare.txt")

# Create a list with all words, Create tuple (word,1), reduce by key i.e. the word
counts = (lines.flatMap(lambda x: x.split(' '))          
                  .map(lambda x: (x, 1))                 
                  .reduceByKey(lambda x,y : x + y))

# get the output on local
output = counts.take(10)                    

# print output
for (word, count) in output:    
    if word.strip() != "":
        print(f"'{word}' occurs {count} times")
# print(counts.toDebugString().decode('utf-8'))<jupyter_output><empty_output><jupyter_text>So that is a small example which counts the number of words in thedocument and prints 10 of them.And most of the work gets done in the second command.Don’t worry if you are not able to follow this yet as I still need totell you about the things that make Spark work.But before we get into Spark basics, Let us refresh some of our PythonBasics. Understanding Spark becomes a lot easier if you have used[functional programming with Python](https://amzn.to/2SuAtzL).For those of you who haven’t used it, below is a brief intro.A functional approach to programming in Python ==============================================![Relatedimage](https://miro.medium.com/proxy/1*nCX6bsSNUF_v2hFKgnaQIA.png)Map ------`map` is used to map a function to an array or alist. Say you want to apply some function to every element in a list.You can do this by simply using a for loop but python lambda functionslet you do this in a single line in Python.<jupyter_code>my_list = [1,2,3,4,5,6,7,8,9,10]

# Lets say I want to square each term in my_list.
squared_list = map(lambda x:x**2, my_list)

print(list(squared_list))<jupyter_output>[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]<jupyter_text>In the above example, you could think of `map`as a function which takes two arguments — A function and a list.It then applies the function to every element of the list.What lambda allows you to do is write an inline function. In here thepart `lambda x:x**2` defines a function that takes x as input and returns x².You could have also provided a proper function in place of lambda. Forexample:<jupyter_code>def squared(x):
    return x**2

my_list = [1,2,3,4,5,6,7,8,9,10]

# Lets say I want to square each term in my_list.
squared_list = map(squared, my_list)

print(list(squared_list))<jupyter_output>[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]<jupyter_text>The same result, but the lambda expressions make the code compact and alot more readable.Filter---------The other function that is used extensively is the `filter` function. This function takes two arguments — A condition and the list to filter.If you want to filter your list using some condition you use`filter`.<jupyter_code>my_list = [1,2,3,4,5,6,7,8,9,10]

# Lets say I want only the even numbers in my list.
filtered_list = filter(lambda x:x%2==0,my_list)
print(list(filtered_list))<jupyter_output>[2, 4, 6, 8, 10]<jupyter_text>Reduce---------The next function I want to talk about is the reduce function. Thisfunction will be the workhorse in Spark.This function takes two arguments — a function to reduce that takes twoarguments, and a list over which the reduce function is to be applied.<jupyter_code>import functools
my_list = [1,2,3,4,5]

# Lets say I want to sum all elements in my list.
sum_list = functools.reduce(lambda x,y:x+y,my_list)
print(sum_list)
import functools
my_list = [1,2,3,4]

# Lets say I want to sum all elements in my list.
sum_list = functools.reduce(lambda x,y:x*y,my_list)
print(sum_list)<jupyter_output>24<jupyter_text>In python2 reduce used to be a part of Python, now we have to use`reduce` as a part of `functools`.Here the lambda function takes in two values x, y and returns their sum.Intuitively you can think that the reduce function works as:```Reduce function first sends 1,2    ; the lambda function returns 3Reduce function then sends 3,3     ; the lambda function returns 6Reduce function then sends 6,4     ; the lambda function returns 10Reduce function finally sends 10,5 ; the lambda function returns 15```A condition on the lambda function we use in reduce is that it must be:-   commutative that is a + b = b + a and-   associative that is (a + b) + c == a + (b + c).In the above case, we used sum which is **commutative as well asassociative**. Other functions that we could have used: `max`**,** `min`, `*` etc.Moving Again to Spark=====================As we have now got the fundamentals of Python Functional Programming outof the way, lets again head to Spark.But first, let us delve a little bit into how spark works. Sparkactually consists of two things a driver and workers.Workers normally do all the work and the driver makes them do that work.RDD---An RDD(Resilient Distributed Dataset) is a parallelized data structurethat gets distributed across the worker nodes. They are the basic unitsof Spark programming.In our wordcount example, in the first line```pylines = sc.textFile("/FileStore/tables/shakespeare.txt")```we took a text file and distributed it across worker nodes so that theycan work on it in parallel. We could also parallelize lists using thefunction `sc.parallelize`For example:<jupyter_code>data = [1,2,3,4,5,6,7,8,9,10]
new_rdd = sc.parallelize(data,4)
new_rdd.collect()<jupyter_output><empty_output><jupyter_text>In Spark, we can do two different types of operations on RDD:Transformations and Actions.1.  **Transformations:** Create new datasets from existing RDDs2.  **Actions:** Mechanism to get results out of SparkTransformation Basics=====================![Image forpost](https://miro.medium.com/max/2560/1*LP9yglc4UeUxDFBoTlfS9w.png)So let us say you have got your data in the form of an RDD.To requote your data is now accessible to the worker machines. You wantto do some transformations on the data now.You may want to filter, apply some function, etc.In Spark, this is done using Transformation functions.Spark provides many transformation functions. You can see acomprehensive list[**here**](http://spark.apache.org/docs/latest/rdd-programming-guide.htmltransformations).Some of the main ones that I use frequently are:Map:-------Applies a given function to an RDD.Note that the syntax is a little bit different from Python, but itnecessarily does the same thing. Don’t worry about `collect` yet. For now, just think of it as a function that collects the data in squared\_rdd back to a list.<jupyter_code>data = [1,2,3,4,5,6,7,8,9,10]
rdd = sc.parallelize(data,4)
squared_rdd = rdd.map(lambda x:x**2)
result_list = squared_rdd.collect()
print(result_list)<jupyter_output>[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]<jupyter_text>Filter:----------Again no surprises here. Takes as input a condition and keeps only thoseelements that fulfill that condition.<jupyter_code>data = [1,2,3,4,5,6,7,8,9,10]
rdd = sc.parallelize(data,4)
filtered_rdd = rdd.filter(lambda x:x%2!=0)
filtered_rdd.collect()<jupyter_output><empty_output><jupyter_text>distinct:------------Returns only distinct elements in an RDD.<jupyter_code>data = [1,2,2,2,2,3,3,3,3,4,5,6,7,7,7,8,8,8,9,10]
rdd = sc.parallelize(data,4)
distinct_rdd = rdd.distinct()
sorted(distinct_rdd.collect())<jupyter_output><empty_output><jupyter_text>flatmap:-----------Similar to `map`, but each input item can bemapped to 0 or more output items.<jupyter_code>data = [1,2,3,4]
rdd = sc.parallelize(data,4)
flat_rdd = rdd.flatMap(lambda x:[x,x**3])
flat_rdd.collect()<jupyter_output><empty_output><jupyter_text>Reduce By Key:-----------------The parallel to the reduce in Hadoop MapReduce.Now Spark cannot provide the value if it just worked with Lists.In Spark, there is a concept of pair RDDs that makes it a lot moreflexible. Let's assume we have a data in which we have a product, itscategory, and its selling price. We can still parallelize the data.<jupyter_code>data = [('Apple','Fruit',200),('Banana','Fruit',24),('Tomato','Fruit',56),('Potato','Vegetable',103),('Carrot','Vegetable',34)]
rdd = sc.parallelize(data,4)
rdd.collect()<jupyter_output><empty_output><jupyter_text>Right now our RDD `rdd` holds tuples.Now we want to find out the total sum of revenue that we got from eachcategory.To do that we have to transform our `rdd` to apair rdd so that it only contains key-value pairs/tuples.<jupyter_code>category_price_rdd = rdd.map(lambda x: (x[1],x[2]))
category_price_rdd.collect()<jupyter_output><empty_output><jupyter_text>Here we used the map function to get it in the format we wanted. Whenworking with textfile, the RDD that gets formed has got a lot ofstrings. We use `map` to convert it into a format that we want.So now our `category_price_rdd` contains theproduct category and the price at which the product sold.Now we want to reduce on the key category and sum the prices. We can dothis by:<jupyter_code>category_total_price_rdd = category_price_rdd.reduceByKey(lambda x,y:x+y)
category_total_price_rdd.collect()<jupyter_output><empty_output><jupyter_text>Group By Key:----------------Similar to `reduceByKey` but does not reducesjust puts all the elements in an iterator. For example, if we wanted tokeep as key the category and as the value all the products we would usethis function.Let us again use `map` to get data in therequired form.<jupyter_code>data = [('Apple','Fruit',200),('Banana','Fruit',24),('Tomato','Fruit',56),('Potato','Vegetable',103),('Carrot','Vegetable',34)]
rdd = sc.parallelize(data,4)
category_product_rdd = rdd.map(lambda x: (x[1],x[0]))
category_product_rdd.collect()<jupyter_output><empty_output><jupyter_text>We then use `groupByKey` as:<jupyter_code>grouped_products_by_category_rdd = category_product_rdd.groupByKey()
findata = grouped_products_by_category_rdd.collect()
for data in findata:
    print(data[0],list(data[1]))<jupyter_output>Fruit ['Apple', 'Tomato', 'Banana']
Vegetable ['Potato', 'Carrot']<jupyter_text>Here the `groupByKey` function worked and itreturned the category and the list of products in that category. Action Basics=============![Image forpost](https://miro.medium.com/max/2560/1*-T8LTnsXH2AhzhbmajacXw.png)You have filtered your data, mapped some functions on it. Done yourcomputation.Now you want to get the data on your local machine or save it to a fileor show the results in the form of some graphs in excel or anyvisualization tool.You will need actions for that. A comprehensive list of actions isprovided[**here**](http://spark.apache.org/docs/latest/rdd-programming-guide.htmlactions)**.**Some of the most common actions that I tend to use are:collect:-----------We have already used this action many times. It takes the whole RDD andbrings it back to the driver program.reduce:----------Aggregate the elements of the dataset using a function func (which takestwo arguments and returns one). The function should be commutative andassociative so that it can be computed correctly in parallel.<jupyter_code>rdd = sc.parallelize([1,2,3,4,5])
rdd.reduce(lambda x,y : x+y)<jupyter_output><empty_output><jupyter_text>take:--------Sometimes you will need to see what your RDD contains without gettingall the elements in memory itself. `take`returns a list with the first n elements of the RDD.<jupyter_code>rdd = sc.parallelize([1,2,3,4,5])
rdd.take(3)<jupyter_output><empty_output><jupyter_text>takeOrdered:---------------`takeOrdered` returns the first n elements ofthe RDD using either their natural order or a custom comparator.<jupyter_code>rdd = sc.parallelize([5,3,12,23])

# descending order
rdd.takeOrdered(3,lambda s:-1*s)
rdd = sc.parallelize([(5,23),(3,34),(12,344),(23,29)])

# descending order
rdd.takeOrdered(3,lambda s:-1*s[1])<jupyter_output><empty_output><jupyter_text>We have our basics covered finally. Let us get back to our wordcountexampleUnderstanding The WordCount Example===================================![Image for post](https://miro.medium.com/max/10368/0*wcxaKwNMIiEsGmW2)Now we sort of understand the transformations and the actions providedto us by Spark.It should not be difficult to understand the wordcount program now. Letus go through the program line by line.The first line creates an RDD and distributes it to the workers.<jupyter_code>lines = sc.textFile("shakespeare/shakespeare.txt")<jupyter_output><empty_output><jupyter_text>This RDD `lines` contains a list of sentences inthe file. You can see the rdd content using `take`<jupyter_code>lines.take(5)<jupyter_output><empty_output><jupyter_text>This RDD is of the form:```py['word1 word2 word3','word4 word3 word2']```This next line is actually the workhorse function in the whole script.<jupyter_code>counts = (lines.flatMap(lambda x: x.split(' '))          
                  .map(lambda x: (x, 1))                 
                  .reduceByKey(lambda x,y : x + y))<jupyter_output><empty_output><jupyter_text>It contains a series of transformations that we do to the lines RDD.First of all, we do a `flatmap` transformation.The `flatmap` transformation takes as input thelines and gives words as output. So after the `flatmap` transformation, the RDD is of the form:```py['word1','word2','word3','word4','word3','word2']```Next, we do a `map` transformation on the`flatmap` output which converts the RDD to :```py[('word1',1),('word2',1),('word3',1),('word4',1),('word3',1),('word2',1)]```Finally, we do a `reduceByKey` transformationwhich counts the number of time each word appeared.After which the RDD approaches the final desirable form.```py[('word1',1),('word2',2),('word3',2),('word4',1)]```This next line is an action that takes the first 10 elements of theresulting RDD locally.<jupyter_code>output = counts.take(10)<jupyter_output><empty_output><jupyter_text>This line just prints the output<jupyter_code>for (word, count) in output:                 
    print("%s: %i" % (word, count))<jupyter_output><empty_output><jupyter_text>And that is it for the wordcount program. Hope you understand it now.So till now, we talked about the Wordcount example and the basictransformations and actions that you could use in Spark. But we don’t dowordcount in real life.We have to work on bigger problems which are much more complex. Worrynot! Whatever we have learned till now will let us do that and more.Spark in Action with Example============================![Image for post](https://miro.medium.com/max/10796/0*e94sY_GitJyJz02J)Let us work with a concrete example which takes care of some usualtransformations.We will work on Movielens[ml-100k.zip](https://github.com/rudrasingh21/Data-ML-100k-/raw/master/ml-100k.zip)dataset which is a stable benchmark dataset. 100,000 ratings from 1000users on 1700 movies. Released 4/1998.Let us start by downloading the data.<jupyter_code># To download the data you would use the following commands:
!rm -rf ml-100k
!wget -P /tmp https://github.com/rudrasingh21/Data-ML-100k-/raw/master/ml-100k.zip
!unzip /tmp/ml-100k.zip -d .
!ls -l ml-100k<jupyter_output><empty_output><jupyter_text>The Movielens dataset contains a lot of files but we are going to beworking with 3 files only:​1) **Users**: This file name is kept as `u.user`. The columns in thisfile are:```py['user_id', 'age', 'sex', 'occupation', 'zip_code']```​2) **Ratings**: This file name is kept as `u.data`. The columns in thisfile are:```py['user_id', 'movie_id', 'rating', 'unix_timestamp']```​3) **Movies**: This file name is kept as `u.item`. The columns in thisfile are:```py['movie_id', 'title', 'release_date', 'video_release_date', 'imdb_url', and 18 more columns.....]```Our business partner now comes to us and asks us to find out the ***25most rated movie titles*** from this data. How many times a movie hasbeen rated?Let us load the data in different RDDs and see what the data contains.<jupyter_code>userRDD = sc.textFile("ml-100k/u.user") 
ratingRDD = sc.textFile("ml-100k/u.data") 
movieRDD = sc.textFile("ml-100k/u.item") 
print("userRDD:",userRDD.take(1))
print("ratingRDD:",ratingRDD.take(1))
print("movieRDD:",movieRDD.take(1))<jupyter_output>userRDD: ['1|24|M|technician|85711']
ratingRDD: ['196\t242\t3\t881250949']
movieRDD: ['1|Toy Story (1995)|01-Jan-1995||http://us.imdb.com/M/title-exact?Toy%20Story%20(1995)|0|0|0|1|1|1|0|0|0|0|0|0|0|0|0|0|0|0|0']<jupyter_text>We note that to answer this question we will need to use the`ratingRDD`. But the `ratingRDD` does not have the movie name.So we would have to merge `movieRDD` and `ratingRDD` using `movie_id`.**How we would do that in Spark?**Below is the code. We also use a new transformation `leftOuterJoin`. Do read the docs and comments in the below code.<jupyter_code># Create a RDD from RatingRDD that only contains the two columns of interest i.e. movie_id,rating.
RDD_movid_rating = ratingRDD.map(lambda x : (x.split("\t")[1],x.split("\t")[2]))
print("RDD_movid_rating:",RDD_movid_rating.take(4))

# Create a RDD from MovieRDD that only contains the two columns of interest i.e. movie_id,title.
RDD_movid_title = movieRDD.map(lambda x : (x.split("|")[0],x.split("|")[1]))
print("RDD_movid_title:",RDD_movid_title.take(2))

# merge these two pair RDDs based on movie_id. For this we will use the transformation leftOuterJoin(). See the transformation document.
rdd_movid_title_rating = RDD_movid_rating.leftOuterJoin(RDD_movid_title)
print("rdd_movid_title_rating:",rdd_movid_title_rating.take(1))

# use the RDD in previous step to create (movie,1) tuple pair RDD
rdd_title_rating = rdd_movid_title_rating.map(lambda x: (x[1][1],1 ))
print("rdd_title_rating:",rdd_title_rating.take(2))

# Use the reduceByKey transformation to reduce on the basis of movie_title
rdd_title_ratingcnt = rdd_title_rating.reduceByKey(lambda x,y: x+y)
print("rdd_title_ratingcnt:",rdd_title_ratingcnt.take(2))

# Get the final answer by using takeOrdered Transformation
print("#####################################")
print("25 most rated movies:",rdd_title_ratingcnt.takeOrdered(25,lambda x:-x[1]))
print("#####################################")<jupyter_output>RDD_movid_rating: [('242', '3'), ('302', '3'), ('377', '1'), ('51', '2')]
RDD_movid_title: [('1', 'Toy Story (1995)'), ('2', 'GoldenEye (1995)')]<jupyter_text>Star Wars is the most rated movie in the Movielens Dataset.Now we could have done all this in a single command using the belowcommand but the code is a little messy now.<jupyter_code>print(((ratingRDD.map(lambda x : (x.split("\t")[1],x.split("\t")[2]))).
     leftOuterJoin(movieRDD.map(lambda x : (x.split("|")[0],x.split("|")[1])))).
     map(lambda x: (x[1][1],1)).
     reduceByKey(lambda x,y: x+y).
     takeOrdered(25,lambda x:-x[1]))<jupyter_output><empty_output><jupyter_text>I did this to show that you can use chaining functions with Spark andyou could bypass the process of variable creation.Let us do one more. For practice:Now we want to find the most highly rated 25 movies using the samedataset. We actually want only those movies which have been rated atleast 100 times.<jupyter_code># We already have the RDD rdd_movid_title_rating: [(u'429', (u'5', u'Day the Earth Stood Still, The (1951)'))]
# We create an RDD that contains sum of all the ratings for a particular movie
rdd_title_ratingsum = (rdd_movid_title_rating.
                        map(lambda x: (x[1][1],int(x[1][0]))).
                        reduceByKey(lambda x,y:x+y))
                        
print("rdd_title_ratingsum:",rdd_title_ratingsum.take(2))
# Merge this data with the RDD rdd_title_ratingcnt we created in the last step
# And use Map function to divide ratingsum by rating count.
rdd_title_ratingmean_rating_count = (rdd_title_ratingsum.
                                    leftOuterJoin(rdd_title_ratingcnt).
                                    map(lambda x:(x[0],(float(x[1][0])/x[1][1],x[1][1]))))
                                    
print("rdd_title_ratingmean_rating_count:",rdd_title_ratingmean_rating_count.take(1))
# We could use take ordered here only but we want to only get the movies which have count
# of ratings more than or equal to 100 so lets filter the data RDD.
rdd_title_rating_rating_count_gt_100 = (rdd_title_ratingmean_rating_count.
                                        filter(lambda x: x[1][1]>=100))
                                        
print("rdd_title_rating_rating_count_gt_100:",rdd_title_rating_rating_count_gt_100.take(1))
# Get the final answer by using takeOrdered Transformation
print("#####################################")
print ("25 highly rated movies:")
print(rdd_title_rating_rating_count_gt_100.takeOrdered(25,lambda x:-x[1][0]))
print("#####################################")<jupyter_output><empty_output><jupyter_text>We have talked about RDDs till now as they are very powerful.You can use RDDs to work with non-relational databases too.They let you do a lot of things that you couldn’t do with SparkSQL?***Yes, you can use SQL with Spark too which I am going to talk aboutnow.***Spark DataFrames================![Image forpost](https://miro.medium.com/max/1234/0*_Xne4_sz6lroaINt.png)Spark has provided DataFrame API to work with relational data. Here is the[documentation](https://docs.databricks.com/spark/latest/dataframes-datasets/introduction-to-dataframes-python.html) for the adventurous folks.Remember that in the background it still is all RDDs and that is why thestarting part of this post focussed on RDDs.I will start with some common functionalities you will need to work withSpark DataFrames. Would look a lot like Pandas with some syntax changes.Reading the File-------------------<jupyter_code>ratings = spark.read.load("ml-100k/u.data",format="csv", sep="\t", inferSchema="true", header="false")<jupyter_output><jupyter_text>Show File------------Here is how we can show files using Spark Dataframes.<jupyter_code>ratings.show(5)
ratings.printSchema()<jupyter_output>root
 |-- user_id: integer (nullable = true)
 |-- movie_id: integer (nullable = true)
 |-- rating: integer (nullable = true)
 |-- unix_timestamp: integer (nullable = true)<jupyter_text>Change Column names----------------------Good functionality. Always required. Don’t forget the `*` in front of the list.<jupyter_code>ratings = ratings.toDF(*['user_id', 'movie_id', 'rating', 'unix_timestamp'])
ratings.show(5)<jupyter_output>+-------+--------+------+--------------+
|user_id|movie_id|rating|unix_timestamp|
+-------+--------+------+--------------+
|    196|     242|     3|     881250949|
|    186|     302|     3|     891717742|
|     22|     377|     1|     878887116|
|    244|      51|     2|     880606923|
|    166|     346|     1|     886397596|
+-------+--------+------+--------------+
only showing top 5 rows<jupyter_text>Some Basic Stats-------------------<jupyter_code>print(ratings.count()) #Row Count
print(len(ratings.columns)) #Column Count<jupyter_output>100000
4<jupyter_text>We can also see the dataframe statistics using:<jupyter_code>ratings.describe().show()<jupyter_output>[Stage 54:>                                                         (0 + 1) / 1]<jupyter_text>Select a few columns-----------------------<jupyter_code>ratings.select('user_id','movie_id').show(5)<jupyter_output>+-------+--------+
|user_id|movie_id|
+-------+--------+
|    196|     242|
|    186|     302|
|     22|     377|
|    244|      51|
|    166|     346|
+-------+--------+
only showing top 5 rows<jupyter_text>Filter---------Filter a dataframe using multiple conditions:<jupyter_code>ratings.filter((ratings.rating==5) & (ratings.user_id==253)).show(5)<jupyter_output>+-------+--------+------+--------------+
|user_id|movie_id|rating|unix_timestamp|
+-------+--------+------+--------------+
|    253|     465|     5|     891628467|
|    253|     510|     5|     891628416|
|    253|     183|     5|     891628341|
|    253|     483|     5|     891628122|
|    253|     198|     5|     891628392|
+-------+--------+------+--------------+
only showing top 5 rows<jupyter_text>Groupby----------We can use groupby function with a spark dataframe too. Pretty much sameas a pandas groupby with the exception that you will need to import`pyspark.sql.functions`<jupyter_code>from pyspark.sql import functions as F
ratings.groupBy("user_id").agg(F.count("user_id"),F.mean("rating")).show(5)<jupyter_output>[Stage 59:>                                                         (0 + 1) / 1]<jupyter_text>Here we have found the count of ratings and average rating from eachuser_id Sort=======<jupyter_code>ratings.sort("user_id").show(5)<jupyter_output>[Stage 62:>                                                         (0 + 1) / 1]<jupyter_text>We can also do a descending sort using `F.desc`function as below.<jupyter_code># descending Sort
from pyspark.sql import functions as F
ratings.sort(F.desc("user_id")).show(5)<jupyter_output><jupyter_text>Joins/Merging with Spark Dataframes===================================We can use SQL with dataframes and thus we can merge dataframes using SQL.Let us try to run some SQL on Ratings.We first register the ratings df to a temporary table ratings\_table onwhich we can run sql operations.As you can see the result of the SQL select statement is again a SparkDataframe.<jupyter_code>ratings.registerTempTable('ratings_table')
newDF = spark.sql('select * from ratings_table where rating > 4')
newDF.show(5)<jupyter_output>+-------+--------+------+--------------+
|user_id|movie_id|rating|unix_timestamp|
+-------+--------+------+--------------+
|    196|     242|     3|     881250949|
|    186|     302|     3|     891717742|
|     22|     377|     1|     878887116|
|    244|      51|     2|     880606923|
|    166|     346|     1|     886397596|
+-------+--------+------+--------------+
only showing top 5 rows<jupyter_text>Let us now add one more Spark Dataframe to the mix to see if we can usejoin using the SQL queries:<jupyter_code># get one more dataframe to join
movies = spark.read.load("ml-100k/u.item",format="csv", sep="|", inferSchema="true", header="false")

# change column names
movies = movies.toDF(*["movie_id","movie_title","release_date","video_release_date","IMDb_URL","unknown","Action","Adventure","Animation ","Children","Comedy","Crime","Documentary","Drama","Fantasy","Film_Noir","Horror","Musical","Mystery","Romance","Sci_Fi","Thriller","War","Western"])

# display
movies.show(5)<jupyter_output><jupyter_text>Now let us try joining the tables on movie\_id to get the name of themovie in the ratings table.<jupyter_code>movies.registerTempTable('movies_table')

spark.sql("""
select 
    ratings_table.*,
    movies_table.movie_title 
from ratings_table 
left join movies_table 
    on movies_table.movie_id = ratings_table.movie_id
""").show(5)<jupyter_output>+-------+--------+------+--------------+--------------------+
|user_id|movie_id|rating|unix_timestamp|         movie_title|
+-------+--------+------+--------------+--------------------+
|    196|     242|     3|     881250949|        Kolya (1996)|
|    186|     302|     3|     891717742|L.A. Confidential...|
|     22|     377|     1|     878887116| Heavyweights (1994)|
|    244|      51|     2|     880606923|Legends of the Fa...|
|    166|     346|     1|     886397596| <NAME> (1997)|
+-------+--------+------+--------------+--------------------+
only showing top 5 rows<jupyter_text>Let us try to do what we were doing earlier with the RDDs. Finding thetop 25 most rated movies:<jupyter_code>mostrateddf = spark.sql("""
select 
    movie_id,
    movie_title, 
    count(user_id) as num_ratings 
from (
    select 
        ratings_table.*,
        movies_table.movie_title 
    from ratings_table 
    left join movies_table 
        on movies_table.movie_id = ratings_table.movie_id
    )A 
group by movie_id, movie_title 
order by num_ratings desc 
""")

mostrateddf.show(5)<jupyter_output>[Stage 72:>                                                         (0 + 1) / 1]<jupyter_text>And finding the top 25 highest rated movies having more than 100 votes:<jupyter_code>highrateddf = spark.sql("""
select 
    movie_id,
    movie_title, 
    avg(rating) as avg_rating,
    count(movie_id) as num_ratings 
from (
    select 
        ratings_table.*,
        movies_table.movie_title 
    from ratings_table 
    left join movies_table 
        on movies_table.movie_id = ratings_table.movie_id
    )A 
group by movie_id, movie_title 
having num_ratings>100 
order by avg_rating desc 
""")

highrateddf.show(5, False)<jupyter_output>[Stage 76:>                                                         (0 + 1) / 1]<jupyter_text>Converting from Spark Dataframe to RDD and vice versa:======================================================Sometimes you may want to convert to RDD from a spark Dataframe or viceversa so that you can have the best of both worlds.To convert from DF to RDD, you can simply do :<jupyter_code>highratedrdd = highrateddf.rdd
highratedrdd.take(2)<jupyter_output><jupyter_text>To go from an RDD to a dataframe:<jupyter_code>from pyspark.sql import Row
# creating a RDD first
data = [('A',1),('B',2),('C',3),('D',4)]
rdd = sc.parallelize(data)

# map the schema using Row.
rdd_rows = rdd.map(lambda x: Row(name=x[0], age=int(x[1])))

# Convert the rdd to Dataframe
df = spark.createDataFrame(rdd_rows)

df.show(5)

df.registerTempTable('people')
spark.sql('select * from people where age > 3').show()<jupyter_output>+----+---+
|name|age|
+----+---+
|   A|  1|
|   B|  2|
|   C|  3|
|   D|  4|
+----+---+

+----+---+
|name|age|
+----+---+
|   D|  4|
+----+---+
{'fraction_non_alphanumeric': 0.14770440251572328, 'fraction_numerical': 0.03411949685534591, 'mean_word_length': 4.052589768033047, 'pattern_counts': {'":': 1, '<': 164, '<?xml version=': 0, '>': 173, 'https://': 12, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 21, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '47', 'dir': 'jupyter-structured-clean-dedup', 'id': '150910', 'license': 'Apache-2.0', 'n_tokens_mistral': 10852, 'n_tokens_neox': 10019, 'n_words': 3218, 'path': 'workspace/spark-introduction.ipynb', 'repo_name': 'sgioldasis/bigdata-infra', 'size': '31806'}