Pyspark Dataframe and SQL

Setup

!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q http://archive.apache.org/dist/spark/spark-3.1.1/spark-3.1.1-bin-hadoop3.2.tgz
!tar xf spark-3.1.1-bin-hadoop3.2.tgz
!pip install -q findspark
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-3.1.1-bin-hadoop3.2"
import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").getOrCreate()
spark

Creating Spark Dataframes

Get Manpreet Buttar’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Pyspark DataFrames are generally preferred over RDDs for most data processing tasks because of High-level abstraction, Optimization, Structured API, Schema inference and enforcement, and Integration with Spark SQL, Support for various data sources:

- SparkSession is the entry point to working with structured data in Spark
- SparkContext is the main entry point for Spark functionality
- With SparkContext, we can create RDDs and perform operations on them

from pyspark.sql import SparkSession # create spark session
spark = SparkSession.builder.getOrCreate()
spark
sc = spark.sparkContext

# Crete DF from RDD
students_data_rdd = sc.textFile('Load file .csv') # Load the data
type(students_data_rdd) # Check the data type
students_data_rdd.collect() # view the data in RDD 
students_data_rdd = students_data_rdd.map(lambda x: x.split(' ')) # Tokenize records
students_data_rdd.collect()
students_data_dataframe = students_data_rdd.toDF(["col1", "col2",....])    # Create dataframe from rdd    
students_data_dataframe.show() # Display dataframe

# Create dataframe from collection
sample_data = [
    (123, "A", "MK",    "Toronto"),
  ...()
]

dataframe_from_collections = spark.createDataFrame(data=sample_data,schema=["roll_no","section","name","city"])
                                                                        
# Display dataframe
dataframe_from_collections.show()
+-------+-------+--------+---------+
|roll_no|section|    name|     city|
+-------+-------+--------+---------+
|    123|      A|      MK|  Toronto|
+-------+-------+--------+---------+
***CSV file
import pyspark.sql.types as tp # Import required libraries
 # Define schema or dataframe
my_schema = tp.StructType([
    tp.StructField(name= "roll_no", dataType= tp.IntegerType()),
    tp.StructField(name= "section", dataType= tp.StringType()),
    tp.StructField(name= "name",    dataType= tp.StringType()),
  ...
])
# Create dataframe from CSV
df_csv_schema = spark.read.csv("file.csv",
                                             header=False, 
                                             schema=my_schema)
# Check type
type(df_csv_schema)
pyspark.sql.dataframe.DataFrame
# Display dataframe
df_csv_schema.show()
+-------+-------+---------+---------+--------+--------+--------+--------+--------+
|roll_no|section|     name|     city|subject1|subject2|subject3|subject4|subject5|
+-------+-------+---------+---------+--------+--------+--------+--------+--------+
|    101|      A|    Rohit|  Gurgaon|      65|      77|      43|      66|      87|
|    102|      B|  Akansha|    Delhi|      55|      46|      24|      66|      77|

+-------+-------+---------+---------+--------+--------+--------+--------+--------+

# Create dataframe with inferSchema
df_csv_infer = spark.read.csv("file.csv",
                                             header=False, 
                                             inferSchema=True)
# Check type
type(df_csv_infer) #pyspark.sql.dataframe.DataFrame
# Display dataframe
df_csv_infer.show()
# 1. Rename columns 
df_csv_infer2 = df_csv_infer.withColumnRenamed("_c0","roll_no")\
                            .withColumnRenamed("_c1","section")\
                        ...
                            .withColumnRenamed("_c8","section_5")
# 2. Rename columns
column_names = ["roll_no","section","section_5"]
# Rename columns
df_csv_infer3 = df_csv_infer.toDF(*column_names)

DataFrames Operations

#Print Schema
data.printSchema() 
#Column Names
healthcare_data.columns 
(healthcare_data.count(), len(healthcare_data.columns)) #Check the Dimensions of the Data
# Select Columns
sample_data = healthcare_data.select("hospital_code",
                                     "department",
                                     "ward_type",
                                     "patient_id",
                                     "age",
                                     "visitors_with_patient")

sample_data.show()
# Drop column
sample_data_without_age = sample_data.drop("age")
sample_data_without_age.show()
# Drop multiple columns
sample_data_drop = sample_data.drop(*["age", "department", "visitors_with_patient"])
sample_data_drop.show()
# Retrieve records with specific column value
sample_data.where(sample_data.columnname==8).show()
sample_data.where((sample_data.col_name==8) & (sample_data.col_name=="value")).show()

# Add New Column
1. Using Expression
# Define expression
country_value = expr("'India'")
# Add new column to dataframe
updated_sample_data = sample_data.withColumn("country", country_value)
# Updated dataframe
updated_sample_data.show()

# Filter condition
filter_expression = expr("IF(ward_type = 'Q', 2, 0)")
updated_sample_data = sample_data.withColumn("ward_number_using_expr", filter_expression)
updated_sample_data.show()

2. Using when & otherwise
# Define the condition
filter_condition = when(sample_data["ward_type"] == 'Q', 2).otherwise(0)
filter_condition = when(sample_data["ward_type"] == 'Q', 2).when(sample_data["visitors_with_patient"] < 4, 1).otherwise(0)

# Add new column
updated_sample_data = sample_data.withColumn("ward_number_using_when", filter_condition)
# show the data
updated_sample_data.show()

3. Using User Defined Function
from pyspark.sql.functions import udf
# Define the function to encode ward_type
def encode_ward_visitor(ward, visitor):
    
    if ward == "Q":
        return 2
    elif visitor < 4:
        return 1
    else:
        return 0
# Convert to udf function
function_with_udf = udf(f= encode_ward_visitor, returnType= tp.IntegerType())
# Create new column
updated_sample_data = sample_data.withColumn("new_column_using_udf", 
                                             function_with_udf(sample_data["ward_type"], 
                                                               sample_data["visitors_with_patient"]))

updated_sample_data.show()

# Sort data
sorted_df = sample_data.orderBy(["hospital_code"])

# Sort in descending order
sorted_df_descending = sample_data.orderBy(["hospital_code"], ascending = False)
sorted_df_multiple = sample_data.orderBy(["hospital_code", "visitors_with_patient"], ascending = [True, False])

# Group by
# Import functions
from pyspark.sql.functions import countDistinct, avg
# Group the data on hospital_code
grouped_data = sample_data.groupBy("hospital_code")
# Average number of visitors_with_patient
grouped_data_average_visitors = grouped_data.agg(avg("visitors_with_patient"))
# Sort grouped data
grouped_data_average_visitors.orderBy(['hospital_code']).show()
grouped_data_department_unique_patient = grouped_data_department.agg(countDistinct("patient_id").alias("unique_patients"))
grouped_data_multiple_columns = grouped_data_department.agg(countDistinct("patient_id").alias("unique_patient"), 
                                                            avg("visitors_with_patient").alias("average_patient_visitors"))

# Inner join
inner_join_df = sample_data.join(case_data, 'case_id')
# Right outer join
right_outer_join_df = sample_data.join(case_data, 'case_id', 'rightouter')

SPARK SQL

# Import required libraries
from pyspark.sql.types import StringType, IntegerType, StructType, StructField, DoubleType
# Define schema
finalSchema = StructType([StructField('age', DoubleType(),True),
                              StructField('workclass',StringType(),True),
                              StructField('fnlwgt', DoubleType(),True),
                              StructField('education', StringType(), True),
                              StructField('education_num', DoubleType(), True)])
# Creating dataframe from the data
df = spark.read.csv('data/MYDATA.csv',schema=finalSchema,header=True)
# Schema of dataframe
df.printSchema()
# Create view for the dataframe
df.createOrReplaceTempView("census")
# View columns
spark.sql("show columns from census").show()
# View select columns
spark.sql("select age, workclass, fnlwgt, education, income from census")

# Create new column
df_mod = spark.sql('select *, fnlwgt%1000 as NEW_COLUMN from census')
# Create view for the dataframe
df_mod.createOrReplaceTempView("census_mod")
# SQL query on new View
spark.sql("select fnlwgt, NEW_COLUMN from census_mod")

# Marital rates by occupation
spark.sql(
  """
  SELECT 
    occupation,
    SUM(1) as n,
    ROUND(AVG(if(LTRIM(marital_status) LIKE 'Married-%',1,0)),2) as married_rate,
    ROUND(AVG(if(lower(marital_status) LIKE '%widow%',1,0)),2) as widow_rate,
    ROUND(AVG(if(LTRIM(marital_status) = 'Divorced',1,0)),2) as divorce_rate,
    ROUND(AVG(if(LTRIM(marital_status) = 'Separated',1,0)),2) as separated_rate,
    ROUND(AVG(if(LTRIM(marital_status) = 'Never-married',1,0)),2) as bachelor_rate
  FROM 
    census_mod 
  GROUP BY 1
  ORDER BY n DESC
  """)

# Convert table to dataframe
df_mod_2 = spark.table("census_mod")
# Columns
df_mod_2.columns

# Average people withh income >50K per occupation
df_result = spark.sql(
  """
  SELECT 
    occupation,
    AVG(IF(income = ' >50K',1,0)) as plus_50k
  FROM 
    census_mod 
  GROUP BY 1
  ORDER BY 2 DESC
  """)
# Display dataframe
df_result.show()

# Convert Spark dataframe to Pandas dataframe
import pandas as pd

# Collect results into a Pandas dataframe
df_pandas = pd.DataFrame(df_result.collect(), columns=df_result.schema.names)
# Display df
df_pandas.head()