Introduction to Google Colab and PySpark
-Table Of Contents:¶
-
-
- Objective -
- Prerequisite -
- Notes from the author -
- Big data, PySpark and Colaboratory -
- Big data -
- PySpark -
- Colaboratory -
- Jupyter notebook basics -
- Code cells -
- Text cells -
- Access to the shell -
- Install Spark -
- Loading Dataset -
- Working with the DataFrame API - -
- Hands-on Questions -
- Functions -
- String Functions -
- Numeric functions -
- Date -
- Working with Dates -
- Working with joins -
- Hands-on again! -
- RDD -
- User-Defined Functions (UDF) -
- Common Questions -
-
-
-
-
-
-
Objective¶
The objective of this notebook is to:
--Give a proper understanding about the different PySpark functions available. -A short introduction to Google Colab, as that is the platform on which this notebook is written on. -
Once you complete this notebook, you should be able to write pyspark programs in an efficent way. The ideal way to use this is by going through the examples given and then trying them on Colab. At the end there are a few hands on questions which you can use to evaluate yourself.
- -Prerequisite¶
- -Although some theory about pyspark and big data will be given in this notebook, I recommend everyone to read more about it and have a deeper understanding on how the functions get executed and the relevance of big data in the current scenario. - A good understanding on python will be an added bonus. -
Notes from the author¶
This tutorial was made using Google Colab so the code you see here is meant to run on a colab notebook.
-It goes through basic PySpark Functions and a short introduction on how to use Colab.
-If you want to view my colab notebook for this particular tutorial, you can view it here. The viewing experience and readability is much better there.
-If you want to try out things with this notebook as a base, feel free to download it from my repo here and then use it with jupyter notebook.
Big data, PySpark and Colaboratory¶
-Big data¶
-Big data usually means data of such huge volume that normal data storage solutions cannot efficently store and process it. In this era, data is being generated at an absurd rate. Data is collected for each movement a person makes. The bulk of big data comes from three primary sources:
--
-
- Social data -
- Machine data -
- Transactional data -
Some common examples for the sources of such data include internet searches, facebook posts, doorbell cams, smartwatches, online shopping history etc. Every action creates data, it is just a matter of of there is a way to collect them or not. But what's interesting is that out of all this data collected, not even 5% of it is being used fully. There is a huge demand for big data professionals in the industry. Even though the number of graduates with a specialization in big data are rising, the problem is that they don't have the practical knowledge about big data scenarios, which leads to bad architecutres and inefficent methods of processing data.
-- -If you are interested to know more about the landscape and technologies involved, here is an article which I found really interesting!
-
PySpark¶
-If you are working in the field of big data, you must have definelty heard of spark. If you look at the Apache Spark website, you will see that it is said to be a Lightning-fast unified analytics engine. PySPark is a flavour of Spark used for processing and analysing massive volumes of data. If you are familiar with python and have tried it for huge datasets, you should know that the execution time can get ridiculous. Enter PySpark!
Imagine your data resides in a distributed manner at different places. If you try brining your data to one point and executing your code there, not only would that be inefficent, but also cause memory issues. Now let's say your code goes to the data rather than the data coming to where your code. This will help avoid unneccesary data movement which will thereby decrease the running time.
-PySpark is the Python API of Spark; which means it can do almost all the things python can. Machine learning(ML) pipelines, exploratory data analysis (at scale), ETLs for data platform, and much more! And all of them in a distributed manner. One of the best parts of pyspark is that if you are already familiar with python, it's really easy to learn.
-Apart from PySpark, there is another language called Scala used for big data processing. Scala is frequently over 10 times faster than Python is native for Hadoop as its based on JVM. But PySpark is getting adopted at a fast rate because of the ease of use, easier learning curve and ML capabilities.
-I will briefly explain how a PySpark job works, but I strongly recommend you read more about the architecture and how everything works. Now, before I get into it, let me talk about some basic jargons first:
-Cluster is a set of loosely or tightly connected computers that work together so that they can be viewed as a single system.
-Hadoop is an open source, scalable, and fault tolerant framework written in Java. It efficiently processes large volumes of data on a cluster of commodity hardware. Hadoop is not only a storage system but is a platform for large data storage as well as processing.
-HDFS (Hadoop distributed file system). It is one of the world's most reliable storage system. HDFS is a Filesystem of Hadoop designed for storing very large files running on a cluster of commodity hardware.
-MapReduce is a data Processing framework, which has 2 phases - Mapper and Reducer. The map procedure performs filtering and sorting, and the reduce method performs a summary operation. It usually runs on a hadoop cluster.
-Transformation refers to the operations applied on a dataset to create a new dataset. Filter, groupBy and map are the examples of transformations.
-Actions Actions refer to an operation which instructs Spark to perform computation and send the result back to driver. This is an example of action.
-Alright! Now that that's out of the way, let me explain how a spark job runs. In simple terma, each time you submit a pyspark job, the code gets internally converted into a MapReduce program and gets executed in the Java virtual machine. Now one of the thoughts that might be popping in your mind will probably be: So the code gets converted into a MapReduce program. Wouldn't that mean MapReduce is faster than pySpark?
Well, the answer is a big NO. This is what makes spark jobs special. Spark is capable of handling a massive amount of data at a time, in it's distributed environment. It does this through in-memory processing, which is what makes it almost 100 times faster than Hadoop. Another factor which amkes it fast is Lazy Evaluation. Spark delays its evaluation as much as it can. Each time you submit a job, spark creates an action plan for how to execute the code, and then does nothing. Finally, when you ask for the result(i.e, calls an action), it executes the plan, which is basically all the transofrmations you have mentioned in your code. That's basically the gist of it.
Now lastly, I want to talk about on more thing. Spark mainly consists of 4 modules:
--
-
- Spark SQL - helps to write spark programs using SQL like queries. -
- Spark Streaming - is an extension of the core Spark API that enables scalable, high-throughput, fault-tolerant stream processing of live data streams. used heavily in processing of social media data. -
- Spark MLLib - is the machine learning component of SPark. It helps train ML models on massive datasets with very high efficeny. -
- Spark GraphX - is the visualization component of Spark. It enables users to view data both as graphs and as collections without data movement or duplication. -
Hopefully this image gives a better idea of what I am talking about:
-
Colaboratory¶
-In the words of Google:
-Colaboratory, or “Colab” for short, is a product from Google Research. Colab allows anybody to write and execute arbitrary python code through the browser, and is especially well suited to machine learning, data analysis and education. More technically, Colab is a hosted Jupyter notebook service that requires no setup to use, while providing free access to computing resources including GPUs.
The reason why I used colab is because of its shareability and free GPU. Yeah you read that right. A FREE GPU! Additionally, it helps use different Google services conveniently. It saves to Google Drive and all the services are very closely related. I recommend you go through the offical overview documentation if you want to know more about it. -If you have more questions about colab, please refer this link
-- -While using a colab notebook, you will need an active internet connection to keep a session alive. If you lose the connection you will have to download the datasets again.
-
Jupyter notebook basics¶
-Code cells¶
-2*3
-import pandas as pd
-print("Hello!")
-Text cells¶
-Hello world!
- -Access to the shell¶
-ls
-pwd
-Install Spark¶
-!apt-get update
-!apt-get install openjdk-8-jdk-headless -qq > /dev/null
-!wget -q http://archive.apache.org/dist/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz
-!tar xf spark-2.4.4-bin-hadoop2.7.tgz
-!pip install -q findspark
-import os
-os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
-os.environ["SPARK_HOME"] = "/content/spark-2.4.4-bin-hadoop2.7"
-!ls
-import findspark
-findspark.init()
-from pyspark import SparkContext
-
-sc = SparkContext.getOrCreate()
-sc
-import pyspark
-from pyspark.sql import SparkSession
-spark = SparkSession.builder.getOrCreate()
-spark
-Loading Dataset¶
-# Downloading and preprocessing Chicago's Reported Crime Data
-!wget https://data.cityofchicago.org/api/views/w98m-zvie/rows.csv?accessType=DOWNLOAD
-!ls
-#Renaming the downloaded file
-!mv rows.csv?accessType=DOWNLOAD reported-crimes.csv
-df = spark.read.csv('reported-crimes.csv',header=True)
-df.show(5)
-Working with the Dataframe API¶
-Viewing Dataframe¶
-In Spark, you have a couple of options to view the DataFrame(DF).
--
-
- take(3) will return a list of three row objects. -
- df.collect() will get all of the data from the entire DataFrame . Be careful when using it, because if you have a large data set when you run collect, you can easily crash the driver node. -
- If you want Spark to print out your DataFrame in a nice format, then try df.show() with the number of rows as paramter. -
N.B: The limit function returns a new DataFrame by taking the first n rows.
- -Schema of a DataFrame¶
-df.dtypes
-df.printSchema()
-# Defining a schema
-from pyspark.sql.types import StructType, StructField, StringType, TimestampType, BooleanType, DoubleType, IntegerType
-df.columns
-labels = [
- ('ID',StringType()),
- ('Case Number',StringType()),
- ('Date',TimestampType()),
- ('Block',StringType()),
- ('IUCR',StringType()),
- ('Primary Type',StringType()),
- ('Description',StringType()),
- ('Location Description',StringType()),
- ('Arrest',StringType()),
- ('Domestic',BooleanType()),
- ('Beat',StringType()),
- ('District',StringType()),
- ('Ward',StringType()),
- ('Community Area',StringType()),
- ('FBI Code',StringType()),
- ('X Coordinate',StringType()),
- ('Y Coordinate',StringType()),
- ('Year',IntegerType()),
- ('Updated On',StringType()),
- ('Latitude',DoubleType()),
- ('Longitude',DoubleType()),
- ('Location',StringType()),
- ('Historical Wards 2003-2015',StringType()),
- ('Zip Codes',StringType()),
- ('Community Areas',StringType()),
- ('Census Tracts',StringType()),
- ('Wards',StringType()),
- ('Boundaries - ZIP Codes',StringType()),
- ('Police Districts',StringType()),
- ('Police Beats',StringType())
-]
-schema = StructType([StructField (x[0], x[1], True) for x in labels])
-schema
-df = spark.read.csv('reported-crimes.csv',schema=schema)
-df.printSchema()
-# The schema comes as we gave!
-df.show()
-# This comes as null which means the datatypes we gave were wrong.
-# So let's just stick with infered schema for now, casting date to date type on the way
-from pyspark.sql.functions import col, to_timestamp
-df = spark.read.csv('reported-crimes.csv',header=True).withColumn('Date',to_timestamp(col('Date'),'MM/dd/yyyy hh:mm:ss a'))
-df.show(5)
-Working with columns¶
-df.Block
-df['Block']
-NOTE:
-- -We can't always use the dot notation because this will break when the column names have reserved names or attributes to the data frame class.
-
df.select(col('Block')).show()
-df.select(df.Block).show()
-df.select('Block','Description').show()
-#Adding a column in PySpark
-# We are adding a column called 'One' at the end
-from pyspark.sql.functions import lit
-df = df.withColumn('One',lit(1))
-df.show(5)
-#Renaming a column in PySpark
-df = df.withColumnRenamed('One', 'Test')
-df.show()
-#Group By a column in PySpark
-df.groupBy('Year').count().show(5)
-#Remove columns in PySpark
-df = df.drop('Test')
-df.show(5)
-Working with Rows¶
-# Filtering rows in PySpark
-df.filter(col('Date')<'2019-06-01').show()
-#Get Unique Rows in PySpark
-df.select('Year').distinct().show()
-# Sort Rows in PySpark
-df.orderBy('Date').show()
-# Append rows in PySpark.
-one_day = spark.read.csv('reported-crimes.csv',header=True).withColumn('Date',to_timestamp(col('Date'),'MM/dd/yyyy hh:mm:ss a')).filter(col('Date')==lit('2019-07-30'))
-df.filter(col('Date')==lit('2019-07-30')).count()
-df.union(one_day).filter(col('Date')==lit('2019-07-30')).count()
-Result:
-- -As you can see here, there were 6 crimes commited on 2019-07-30, and after union, there's 12 records.
-
# Top 10 number of reported crimes by Primary Type, in descending order of Occurence
-df.groupBy("Primary Type").count().orderBy('count',ascending=False).show(10)
-Hands-on Questions 🤚 !¶
-What percentage of reported crimes resulted in an arrest?
- -# Answer
-What are the top 3 locations for reported crimes?
- -# Answer
-Functions¶
-# Functions available in PySpark
-from pyspark.sql import functions
-print(dir(functions))
-String Functions¶
-# Loading the data
-from pyspark.sql.functions import col
-df = spark.read.csv('reported-crimes.csv',header=True).withColumn('Date',to_timestamp(col('Date'),'MM/dd/yyyy hh:mm:ss a'))
-Display the Primary Type column in lower and upper characters, and the first 4 characters of the column
- -from pyspark.sql.functions import col,lower, upper, substring
-help(substring)
-df.select(lower(col('Primary Type')),upper(col('Primary Type')),substring(col('Primary Type'),1,4)).show(5)
-Numeric functions¶
-Show the oldest date and the most recent date
- -from pyspark.sql.functions import min, max
-df.select(min(col('Date')), max(col('Date'))).show()
-Date¶
-What is 3 days earlier that the oldest date and 3 days later than the most recent date?
- -from pyspark.sql.functions import date_add, date_sub
-df.select(date_add(max(col('Date')),3), date_sub(min(col('Date')),3)).show()
-Working with Dates¶
-from pyspark.sql.functions import to_date, to_timestamp, lit
-df = spark.createDataFrame([('2019-12-25 13:30:00',)], ['Christmas'])
-df.show()
-df.select(to_date(col('Christmas'),'yyyy-MM-dd HH:mm:ss'), to_timestamp(col('Christmas'),'yyyy-MM-dd HH:mm:ss')).show()
-df = spark.createDataFrame([('25/Dec/2019 13:30:00',)], ['Christmas'])
-df.select(to_date(col('Christmas'),'dd/MMM/yyyy HH:mm:ss'), to_timestamp(col('Christmas'),'dd/MMM/yyyy HH:mm:ss')).show()
-df = spark.createDataFrame([('12/25/2019 01:30:00 PM',)], ['Christmas'])
-df.show(1)
-df.show(1, truncate = False)
-df.select(to_date(col('Christmas'),'MM/dd/yyyy hh:mm:ss aa'), to_timestamp(col('Christmas'),'MM/dd/yyyy hh:mm:ss aa')).show()
-Working with joins¶
-# Loading the data
-from pyspark.sql.functions import col
-df = spark.read.csv('reported-crimes.csv',header=True).withColumn('Date',to_timestamp(col('Date'),'MM/dd/yyyy hh:mm:ss a'))
-# Dowloading police station data
-!wget -O police-station.csv https://data.cityofchicago.org/api/views/z8bn-74gv/rows.csv?accessType=DOWNLOAD
-ps = spark.read.csv("police-station.csv", header=True)
-ps.show()
-The reported crimes dataset has only the district number. Add the district name by joining with the police station dataset.
- -# Caching the crimes dataset to speed things up, and then, since spark does lazy evaluation, gonna run an action to make it evaluated.
-df.cache()
-df.count()
-ps.select(col('DISTRICT')).distinct().show()
-df.select(col('District')).distinct().show()
-# Transfomring crime data to remove 0 from beginning, inroder to match the data
-from pyspark.sql.functions import lpad
-ps = ps.withColumn('Format_district',lpad(col('DISTRICT'),3,'0'))
-ps.select(col('Format_district')).show()
-# Executing the join and deleting some column so that we don't get too much data
-df.join(ps, df.District == ps.Format_district, 'left_outer').drop(
- 'ADDRESS',
- 'CITY',
- 'STATE',
- 'ZIP',
- 'WEBSITE',
- 'PHONE',
- 'FAX',
- 'TTY',
- 'X COORDINATE',
- 'Y COORDINATE',
- 'LATITUDE',
- 'LONGITUDE',
- 'LOCATION').show()
-Hands-on again!¶
-What is the most frequently reported non-criminal activity?
- -
-Find the day of the week with the most reported crime?
- -
-Using a bar chart, plot which day of the week has the most number of reported crime.
- -from pyspark.sql.functions import date_format, col
-dow = [x[0] for x in df.groupBy(date_format(col('Date'),'E')).count().collect()]
-print(dow)
-cnt = [x[1] for x in df.groupBy(date_format(col('Date'),'E')).count().collect()]
-print(cnt)
-import pandas as pd
-import matplotlib.pyplot as plt
-
-cp = pd.DataFrame({'Day_of_week':dow, 'Count':cnt})
-cp.head()
-cp.sort_values('Count', ascending=False).plot(kind='bar', color= 'red', x='Day_of_week', y='Count')
-plt.xlabel("Day of week")
-plt.ylabel("Number of reported crimes")
-plt.title("No.of reported crimes per day")
-
-RDD¶
--With map, you define a function and then apply it record by record. Flatmap returns a new RDD by first applying a function to all of the elements in RDDs and then flattening the result. Filter, returns a new RDD. Meaning only the elements that satisfy a condition. With reduce, we are taking neighboring elements and producing a single combined result. -For example, let's say you have a set of numbers. You can reduce this to its sum by providing a function that takes as input two values and reduces them to one.
-
Some of the reasons you would use a dataframe over RDD are:
--
-
- It's ability to represnt data as rows and columns. But this also means it can only hold structred and semi-structured data. -
- It allows processing data in different formats (AVRO, CSV, JSON, and storage system HDFS, HIVE tables, MySQL). -
- It's superior job Optimization capability. -
- DataFrame API is very easy to use. - -
psrdd = sc.textFile('police-station.csv')
-print(psrdd.first())
-ps_header = psrdd.first()
-ps_rest = psrdd.filter(lambda line: line!=ps_header)
-print(ps_rest.first())
-How many police stations are there?
- -ps_rest.map(lambda line: line.split(",")).count()
-Display the District ID, District name, Address and Zip for the police station with District ID 7
- -# District is column 0
-(ps_rest.filter(lambda line: line.split(",")[0]=='7').
- map(lambda line: (line.split(",")[0],
- line.split(",")[1],
- line.split(",")[2],
- line.split(",")[5])).collect())
-Police stations 10 and 11 are geographically close to each other. Display the District ID, District name, address and zip code
- -# District is column 0
-(ps_rest.filter(lambda line: line.split(",")[0] in ['10', '11']).
- map(lambda line: (line.split(",")[0],
- line.split(",")[1],
- line.split(",")[2],
- line.split(",")[5])).collect())
-User-Defined Functions (UDF)¶
-PySpark User-Defined Functions (UDFs) help you convert your python code into a scalable version of itself. It comes in handy more than you can imagine, but beware, as the performance is less when you compare it with pyspark functions. You can view examples of how UDF works here. What I will give in this section is some theory on how it works, and why it is slower.
-When you try to run a UDF in PySpark, each executor creates a python process. Data will be serialised and deserialised between each executor and python. This leads to lots of performance impact and overhead on spark jobs, making it less efficent than using spark dataframes. Apart from this, sometimes you might have memory issues while using UDFs. The Python worker consumes huge off-heap memory and so it often leads to memoryOverhead, thereby failing your job. Keeping these in mind, I wouldn't recommend using them, but at the end of the day, your choice.
- -Common Questions¶
-Recommeded IDE¶
I personally prefer PyCharm while coding in Python/PySpark. It's based on IntelliJ IDEA so it has a lot of features! And the main advantage I have felt is the ease of installing pyspark and other packages. You can customize it with themes and plugins, and it lets you enhance productivity while coding by providing some features like suggestions, Local VCS etc.
- -Submitting a spark job¶
The python syntax for running jobs is: python <file_name>.py <arg1> <arg2> ...
-But when you submit a spark job you have to use spark-submit to run the application.
Here is a simple example of a spark-submit command:
-spark-submit filename.py --named_argument 'arguemnt value'
-Here, named_argument is an arguemnt that you are reading from inside your script.
There are other options you can pass in the command, like:
---py-files which helps you pass a python file to read in your file,
---files which helps pass other files like txt or config,
---deploy-mode which tells wether to deploy your worker node on cluster or locally
---conf which helps pass different configurations, like memoryOverhead, dynamicAllocation etc.
There is an entire page in spark documentation dedicated to this. I highly recommend you go through it once.
- -