Before diving into the equivalents, we first need to set the floor for later. It goes without saying that the first step is to import the needed libraries:
import pandas as pd
import pyspark.sql.functions as F
The entry point into PySpark functionalities is the SparkSession class. Through the SparkSession instance, you can create dataframes, apply all kinds of transformations, read and write files, etc … To define a SparkSession you can use the following :
from pyspark.sql import SparkSession
spark = SparkSession\
.builder\
.appName('SparkByExamples.com')\
.getOrCreate()
Now that everything is set, let’s jump right into the Pandas vs PySpark part!
First, let’s define a data sample we’ll be using:
columns = ["employee","department","state","salary","age"]
data = [("Alain","Sales","Paris",60000,34),
("Ahmed","Sales","Lyon",80000,45),
("Ines","Sales","Nice",55000,30),
("Fatima","Finance","Paris",90000,28),
("Marie","Finance","Nantes",100000,40)]
To create a Pandas DataFrame , we can use the following:
df = pd.DataFrame(data=data, columns=columns)# Show a few lines
df.head(2)
PySpark
df = spark.createDataFrame(data).toDF(*columns)# Show a few lines
df.limit(2).show()
Pandas
types_dict = {
"employee": pd.Series([r[0] for r in data], dtype='str'),
"department": pd.Series([r[1] for r in data], dtype='str'),
"state": pd.Series([r[2] for r in data], dtype='str'),
"salary": pd.Series([r[3] for r in data], dtype='int'),
"age": pd.Series([r[4] for r in data], dtype='int')
}df = pd.DataFrame(types_dict)
You can check your types by executing this line:
df.dtypes
PySpark
from pyspark.sql.types import StructType,StructField, StringType, IntegerTypeschema = StructType([ \
StructField("employee",StringType(),True), \
StructField("department",StringType(),True), \
StructField("state",StringType(),True), \
StructField("salary", IntegerType(), True), \
StructField("age", IntegerType(), True) \
])
df = spark.createDataFrame(data=data,schema=schema)
You can check your DataFrame’s schema by executing :
df.dtypes
# OR
df.printSchema()
Reading and writing are so similar in Pandas and PySpark. The syntax is the following for each: Pandas
df = pd.read_csv(path, sep=';', header=True)
df.to_csv(path, ';', index=False)
PySpark
df = spark.read.csv(path, sep=';')
df.coalesce(n).write.mode('overwrite').csv(path, sep=';')
Note 1 💡: that you can specify a column on which you wish to partition:
df.partitionBy("department","state").write.mode('overwrite').csv(path, sep=';')
Note 2 💡: You can read and write in different formats like parquet format by changing the CSV by parquet in all the lines of code above
Specific columns
Selecting certain columns in Pandas is done like so: Pandas
columns_subset = ['employee', 'salary']df[columns_subset].head()
df.loc[:, columns_subset].head()
Whereas in PySpark, we need to use the select method with a list of columns: PySpark
columns_subset = ['employee', 'salary']df.select(columns_subset).show(5)
Specific lines
To select a range of lines, you can use the ilocmethod in Pandas:
Pandas
# Take a sample ( first 2 lines )df.iloc[:2].head()
In Spark, it is not possible to get any range of line numbers. It is however possible to select the n first lines like so:
PySpark
df.take(2).head()
# Or
df.limit(2).head()
Note 💡 : With spark keep in mind the data is potentially distributed over different compute nodes and the “first” lines may change from run to run since there is no underlying order
Using a condition
It is possible to filter data based on a certain condition. The syntax in Pandas is the following:
Pandas
# First method
flt = (df['salary'] >= 90_000) & (df['state'] == 'Paris')
filtered_df = df[flt]# Second Method: Using query which is generally faster
filtered_df = df.query('(salary >= 90_000) and (state == "Paris")')
# Or
target_state = "Paris"
filtered_df = df.query('(salary >= 90_000) and (state == @target_state)')
In Spark, the same result can be found by using the filtermethod or executing an SQL query. The syntax is the following:
PySpark
# First method
filtered_df = df.filter((F.col('salary') >= 90_000) & (F.col('state') == 'Paris'))# Second Method:
df.createOrReplaceTempView("people")
filtered_df = spark.sql("""
SELECT * FROM people
WHERE (salary >= 90000) and (state == "Paris")
""")
# OR
filtered_df = df.filter(F.expr('(salary >= 90000) and (state == "Paris")'))
In Pandas, there are several ways to add a column:
Pandas
seniority = [3, 5, 2, 4, 10]
# Method 1
df['seniority'] = seniority# Method 2
df.insert(2, "seniority", seniority, True)
In PySpark there is a specific method called withColumn that can be used to add a column:
PySpark
seniority = [3, 5, 2, 4, 10]
df = df.withColumn('seniority', seniority)
Two Dataframes
Pandas
df_to_add = pd.DataFrame(data=[("Robert","Advertisement","Paris",55000,27)], columns=columns)
df = pd.concat([df, df_to_add], ignore_index = True)
PySpark
df_to_add = spark.createDataFrame([("Robert","Advertisement","Paris",55000,27)]).toDF(*columns)
df = df.union(df_to_add)
Multiple Dataframes
Pandas
dfs = [df, df1, df2,...,dfn]
df = pd.concat(dfs, ignore_index = True)
The method unionAllof PySpark only concatenates two dataframes. A workaround for this limit is to iterate the concatenations as many times as needed. For a more compact elegant syntax, we’re going to avoid loops and use the reduce method to apply unionAll:
PySpark
from functools import reduce
from pyspark.sql import DataFramedef unionAll(*dfs):
return reduce(DataFrame.unionAll, dfs)
dfs = [df, df1, df2,...,dfn]
df = unionAll(*dfs)
Computing specified statistics
In some cases, we need to perform some data analysis through some statistical KPIs. Both Pandas and PySpark offer the possibility to get very easily the following pieces of information for each column in the dataframe:
- The count of the column elements
- The mean of the column elements
- The stddev
- The min value
- Three percentiles: 25%, 50%, and 75%
- The max value
You can compute these values simply by executing one of these lines:
Pandas and PySpark
df.summary()
#OR
df.describe() # the method describe doesn't return the percentiles
Aggregations
To perform some aggregations, the syntax is almost the Pandas and PySpark: Pandas
df.groupby('department').agg({'employee': 'count', 'salary':'max', 'age':'mean'})
PySpark
df.groupBy('department').agg({'employee': 'count', 'salary':'max', 'age':'mean'})
However, the results need some tweaking to be similar in pandas and PySpark. 1. In pandas the column to group by becomes the index:
To get it back as a column, we need to apply the reset_index method: Pandas
df.groupby('department').agg({'employee': 'count', 'salary':'max', 'age':'mean'}).reset_index()
- In PySpark the names of the columns get modified in the resulting dataframe, mentioning the performed aggregation:
If you wish to avoid this, you’ll need to use the alias method like so:
df.groupBy('department').agg(F.count('employee').alias('employee'), F.max('salary').alias('salary'), F.mean('age').alias('age'))
Apply a transformation over a column
To apply a certain transformation over a column, the apply method is no longer an option in PySpark. Instead, we can use a method called udf ( or user-defined function) that envelopes a python function.
For example, we need to increase salary by 15% if the salary is under 60000 and by 5% if over 60000.
The syntax in pandas is the following:
df['new_salary'] = df['salary'].apply(lambda x: x*1.15 if x<= 60000 else x*1.05)
The equivalent in PySpark is the following:
from pyspark.sql.types import FloatTypedf.withColumn('new_salary', F.udf(lambda x: x*1.15 if x<= 60000 else x*1.05, FloatType())('salary'))
⚠️ Note that the udfmethod needs the data type to be specified explicitly (in our case FloatType)
Final thoughts
To conclude, it is clear that there are a lot of similarities between the syntax of Pandas and PySpark. This should facilitate immensely the transition from one to another.
Using PySpark can give you a great advantage when working with big datasets since it allows parallel computing. However, if the dataset you’re working with is small, it becomes quickly more efficient to revert to the one and only Pandas.
Since this article is all about transitioning smoothly from Pandas to PySpark, it is important to mention that there is a pandas equivalent API called Koalas that works on Apache Spark and therefore fills this gap between the two.
Thank you for sticking this far. Stay safe and see you in the next story 😊!
Before diving into the equivalents, we first need to set the floor for later. It goes without saying that the first step is to import the needed libraries:
import pandas as pd
import pyspark.sql.functions as F
The entry point into PySpark functionalities is the SparkSession class. Through the SparkSession instance, you can create dataframes, apply all kinds of transformations, read and write files, etc … To define a SparkSession you can use the following :
from pyspark.sql import SparkSession
spark = SparkSession\
.builder\
.appName('SparkByExamples.com')\
.getOrCreate()
Now that everything is set, let’s jump right into the Pandas vs PySpark part!
First, let’s define a data sample we’ll be using:
columns = ["employee","department","state","salary","age"]
data = [("Alain","Sales","Paris",60000,34),
("Ahmed","Sales","Lyon",80000,45),
("Ines","Sales","Nice",55000,30),
("Fatima","Finance","Paris",90000,28),
("Marie","Finance","Nantes",100000,40)]
To create a Pandas DataFrame , we can use the following:
df = pd.DataFrame(data=data, columns=columns)# Show a few lines
df.head(2)
PySpark
df = spark.createDataFrame(data).toDF(*columns)# Show a few lines
df.limit(2).show()
Pandas
types_dict = {
"employee": pd.Series([r[0] for r in data], dtype='str'),
"department": pd.Series([r[1] for r in data], dtype='str'),
"state": pd.Series([r[2] for r in data], dtype='str'),
"salary": pd.Series([r[3] for r in data], dtype='int'),
"age": pd.Series([r[4] for r in data], dtype='int')
}df = pd.DataFrame(types_dict)
You can check your types by executing this line:
df.dtypes
PySpark
from pyspark.sql.types import StructType,StructField, StringType, IntegerTypeschema = StructType([ \
StructField("employee",StringType(),True), \
StructField("department",StringType(),True), \
StructField("state",StringType(),True), \
StructField("salary", IntegerType(), True), \
StructField("age", IntegerType(), True) \
])
df = spark.createDataFrame(data=data,schema=schema)
You can check your DataFrame’s schema by executing :
df.dtypes
# OR
df.printSchema()
Reading and writing are so similar in Pandas and PySpark. The syntax is the following for each: Pandas
df = pd.read_csv(path, sep=';', header=True)
df.to_csv(path, ';', index=False)
PySpark
df = spark.read.csv(path, sep=';')
df.coalesce(n).write.mode('overwrite').csv(path, sep=';')
Note 1 💡: that you can specify a column on which you wish to partition:
df.partitionBy("department","state").write.mode('overwrite').csv(path, sep=';')
Note 2 💡: You can read and write in different formats like parquet format by changing the CSV by parquet in all the lines of code above
Specific columns
Selecting certain columns in Pandas is done like so: Pandas
columns_subset = ['employee', 'salary']df[columns_subset].head()
df.loc[:, columns_subset].head()
Whereas in PySpark, we need to use the select method with a list of columns: PySpark
columns_subset = ['employee', 'salary']df.select(columns_subset).show(5)
Specific lines
To select a range of lines, you can use the ilocmethod in Pandas:
Pandas
# Take a sample ( first 2 lines )df.iloc[:2].head()
In Spark, it is not possible to get any range of line numbers. It is however possible to select the n first lines like so:
PySpark
df.take(2).head()
# Or
df.limit(2).head()
Note 💡 : With spark keep in mind the data is potentially distributed over different compute nodes and the “first” lines may change from run to run since there is no underlying order
Using a condition
It is possible to filter data based on a certain condition. The syntax in Pandas is the following:
Pandas
# First method
flt = (df['salary'] >= 90_000) & (df['state'] == 'Paris')
filtered_df = df[flt]# Second Method: Using query which is generally faster
filtered_df = df.query('(salary >= 90_000) and (state == "Paris")')
# Or
target_state = "Paris"
filtered_df = df.query('(salary >= 90_000) and (state == @target_state)')
In Spark, the same result can be found by using the filtermethod or executing an SQL query. The syntax is the following:
PySpark
# First method
filtered_df = df.filter((F.col('salary') >= 90_000) & (F.col('state') == 'Paris'))# Second Method:
df.createOrReplaceTempView("people")
filtered_df = spark.sql("""
SELECT * FROM people
WHERE (salary >= 90000) and (state == "Paris")
""")
# OR
filtered_df = df.filter(F.expr('(salary >= 90000) and (state == "Paris")'))
In Pandas, there are several ways to add a column:
Pandas
seniority = [3, 5, 2, 4, 10]
# Method 1
df['seniority'] = seniority# Method 2
df.insert(2, "seniority", seniority, True)
In PySpark there is a specific method called withColumn that can be used to add a column:
PySpark
seniority = [3, 5, 2, 4, 10]
df = df.withColumn('seniority', seniority)
Two Dataframes
Pandas
df_to_add = pd.DataFrame(data=[("Robert","Advertisement","Paris",55000,27)], columns=columns)
df = pd.concat([df, df_to_add], ignore_index = True)
PySpark
df_to_add = spark.createDataFrame([("Robert","Advertisement","Paris",55000,27)]).toDF(*columns)
df = df.union(df_to_add)
Multiple Dataframes
Pandas
dfs = [df, df1, df2,...,dfn]
df = pd.concat(dfs, ignore_index = True)
The method unionAllof PySpark only concatenates two dataframes. A workaround for this limit is to iterate the concatenations as many times as needed. For a more compact elegant syntax, we’re going to avoid loops and use the reduce method to apply unionAll:
PySpark
from functools import reduce
from pyspark.sql import DataFramedef unionAll(*dfs):
return reduce(DataFrame.unionAll, dfs)
dfs = [df, df1, df2,...,dfn]
df = unionAll(*dfs)
Computing specified statistics
In some cases, we need to perform some data analysis through some statistical KPIs. Both Pandas and PySpark offer the possibility to get very easily the following pieces of information for each column in the dataframe:
- The count of the column elements
- The mean of the column elements
- The stddev
- The min value
- Three percentiles: 25%, 50%, and 75%
- The max value
You can compute these values simply by executing one of these lines:
Pandas and PySpark
df.summary()
#OR
df.describe() # the method describe doesn't return the percentiles
Aggregations
To perform some aggregations, the syntax is almost the Pandas and PySpark: Pandas
df.groupby('department').agg({'employee': 'count', 'salary':'max', 'age':'mean'})
PySpark
df.groupBy('department').agg({'employee': 'count', 'salary':'max', 'age':'mean'})
However, the results need some tweaking to be similar in pandas and PySpark. 1. In pandas the column to group by becomes the index:
To get it back as a column, we need to apply the reset_index method: Pandas
df.groupby('department').agg({'employee': 'count', 'salary':'max', 'age':'mean'}).reset_index()
- In PySpark the names of the columns get modified in the resulting dataframe, mentioning the performed aggregation:
If you wish to avoid this, you’ll need to use the alias method like so:
df.groupBy('department').agg(F.count('employee').alias('employee'), F.max('salary').alias('salary'), F.mean('age').alias('age'))
Apply a transformation over a column
To apply a certain transformation over a column, the apply method is no longer an option in PySpark. Instead, we can use a method called udf ( or user-defined function) that envelopes a python function.
For example, we need to increase salary by 15% if the salary is under 60000 and by 5% if over 60000.
The syntax in pandas is the following:
df['new_salary'] = df['salary'].apply(lambda x: x*1.15 if x<= 60000 else x*1.05)
The equivalent in PySpark is the following:
from pyspark.sql.types import FloatTypedf.withColumn('new_salary', F.udf(lambda x: x*1.15 if x<= 60000 else x*1.05, FloatType())('salary'))
⚠️ Note that the udfmethod needs the data type to be specified explicitly (in our case FloatType)
Final thoughts
To conclude, it is clear that there are a lot of similarities between the syntax of Pandas and PySpark. This should facilitate immensely the transition from one to another.
Using PySpark can give you a great advantage when working with big datasets since it allows parallel computing. However, if the dataset you’re working with is small, it becomes quickly more efficient to revert to the one and only Pandas.
Since this article is all about transitioning smoothly from Pandas to PySpark, it is important to mention that there is a pandas equivalent API called Koalas that works on Apache Spark and therefore fills this gap between the two.
Thank you for sticking this far. Stay safe and see you in the next story 😊!