Explain the query processing in PySpark

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[Department#1], functions=[sum(Salary#2L)])
+- Exchange hashpartitioning(Department#1, 200),
ENSURE_REQUIREMENTS, [id=#60]
+- HashAggregate(keys=[Department#1],
functions=[partial_sum(Salary#2L)])
+- InMemoryTableScan [Department#1, Salary#2L]
+- InMemoryRelation [Name#0, Department#1, Salary#2L], StorageLevel(disk, memory, deserialized, 1 replicas)
+- *(1) Scan ExistingRDD[Name#0,Department#1,Salary#2L]


Let's break down what the physical plan is showing you:

1. **AdaptiveSparkPlan:**

The top node, `AdaptiveSparkPlan isFinalPlan=false`, indicates that Spark's Adaptive Query Execution (AQE) is enabled. AQE means Spark can adjust its physical plan at runtime based on the actual data and statistics. Here, it informs you that the current plan is not final and may be optimized further as the job executes.

2. **Final Global Aggregation (HashAggregate):**

The next step is a `HashAggregate` node that groups data by the key `[Department#1]` and uses the aggregation function `sum(Salary#2L)`. This is the final step in computing the total salary per department. Because no grouping keys are passed to the final `groupBy()` (if you had one on the top level, it would be a global aggregation), here it's grouping on the department key to produce the desired result.

3. **Data Exchange (Exchange):**

Before reaching the final aggregation, there's an `Exchange` node. This node handles the data shuffle by redistributing rows across 200 partitions based on hash partitioning of the department column. The exchange ensures that all rows for the same department end up in the same partition so that the subsequent aggregation can compute the final sum correctly. The `ENSURE_REQUIREMENTS` note indicates that Spark is satisfying physical properties (like partitioning) required by the following operators.

4. **Partial Aggregation (HashAggregate):**

Beneath the exchange, another `HashAggregate` node appears. This node computes partial sums of salaries per department. Partial aggregation is a common technique in distributed computing because it reduces the amount of data that has to be shuffled over the network by performing some of the aggregation locally within each partition.

5. **Data Source (InMemoryTableScan and InMemoryRelation):**

The data is ultimately sourced from an `InMemoryTableScan` on the columns `[Department#1, Salary#2L]`. This scan reads data from an `InMemoryRelation`, which is a cached version of your dataset stored in memory (with a storage level that includes disk fallback and a single replica). The `Scan ExistingRDD` at the bottom indicates that this cached DataFrame (or RDD) is being scanned to provide the required columns to the aggregation pipeline.

In summary, the plan shows that Spark is reading data from memory, performing a local (partial) aggregation to compute partial sums of salaries by department, then shuffling the data so that all rows with the same department are grouped together, and finally computing the global sum for each department. This multi-phase strategy (partial then final aggregation) is used to optimize performance and reduce data movement across nodes.

How to flatten a complex JSON file - Example 2

from pyspark.sql import SparkSession
from pyspark.sql.types import ArrayType, StructType
from pyspark.sql.functions import col, explode_outer

def flatten(df):

""" Recursively flattens a PySpark DataFrame with nested structures. For any column whose type is either ArrayType or StructType: - If it is a StructType, the function expands each field of the struct into a new column. The new column names are in the form "parentField_childField". - If it is an ArrayType, the function uses `explode_outer` to convert each element of the array into a separate row (which is useful for arrays of structs or primitive types).

Parameters: df (DataFrame): The input DataFrame that may contain nested columns.

Returns: DataFrame: A flattened DataFrame with no nested columns. """

# Identify columns whose type is either ArrayType or StructType.

complex_fields = {field.name: field.dataType for field in df.schema.fields if isinstance(field.dataType, (ArrayType, StructType))}

while complex_fields:
for col_name, col_type in complex_fields.items():
if isinstance(col_type, StructType):
# For a struct: expand its fields as separate columns.
expanded = [ col(col_name + '.' + subfield.name).alias(col_name + '_' + subfield.name) for subfield in col_type.fields ]

df = df.select("*", *expanded).drop(col_name)
elif isinstance(col_type, ArrayType):
# For an array, explode it so that each element becomes a separate row.
df = df.withColumn(col_name, explode_outer(col_name))
# Recompute the schema to check whether more nested columns remain.

complex_fields = {field.name: field.dataType for field in df.schema.fields if isinstance(field.dataType, (ArrayType, StructType))}

return df
Example Usage
if __name__ == "__main__":
spark = SparkSession.builder.appName("FlattenNestedJson").getOrCreate()

# Replace this with the path to your nested employee JSON file.
json_file_path = "/path/to/employee_record.json"

# Read the nested JSON file.
df = spark.read.json(json_file_path)
# Apply the flatten function.
flat_df = flatten(df)
# Display the flattened DataFrame.
flat_df.show(truncate=False)
spark.stop()



Detecting Complex Types: The function first builds a dictionary (complex_fields) of column names pointing to their data types for any field that is a nested structure (either an array or a struct).

Processing Structs: For each field of type StructType, the code iterates over its subfields and creates new columns named in the pattern "parent_subfield". The original nested column is then dropped.

Processing Arrays: For columns with ArrayType, the function calls explode_outer which converts each element of the array into a separate row (ensuring that null or empty arrays are handled gracefully).

Iterative Flattening: After processing the current set of nested fields, the function rebuilds the dictionary to catch any newly exposed nested fields. This loop continues until no more complex types remain.

How to flatten a complex JSON file - Example 1

from pyspark.sql.types import *
from pyspark.sql.functions import *

def flatten(df):
compute Complex Fields (Lists and Structs) in Schema
complex_fields = dict([(field.name, field.dataType)
for field in df.schema.fields
if type(field.dataType) == ArrayType or type(field.dataType) == StructType])

while len(complex_fields)!=0:

col_name=list(complex_fields.keys())[0]
print ("Processing :"+col_name+" Type : "+str(type(complex_fields[col_name])))

if StructType then convert all sub element to columns.
i.e. flatten structs
if (type(complex_fields[col_name]) == StructType):
expanded = [col(col_name+'.'+k).alias(col_name+'_'+k) for k in [ n.name for n in complex_fields[col_name]]]

df=df.select("*", *expanded).drop(col_name)

if ArrayType then add the Array Elements as Rows using the explode function

i.e. explode Arrays
elif (type(complex_fields[col_name]) == ArrayType):

df=df.withColumn(col_name,explode_outer(col_name))

recompute remaining Complex Fields in Schema
complex_fields = dict([(field.name, field.dataType)
for field in df.schema.fields
if type(field.dataType) == ArrayType or type(field.dataType) == StructType])
return df


Sample Nested Employee Data

[
{
"employeeId": "E001",
"name": {
"first": "John",
"last": "Doe"
},
"contact": {
"email": "john.doe@example.com",
"phones": [
"555-1234",
"555-5678"
]
},
"address": {
"street": "123 Elm St",
"city": "Springfield",
"state": "IL",
"zipcode": "62704"
},
"department": {
"deptId": "D001",
"deptName": "Engineering"
},
"projects": [
{ "projectId": "P001",
"projectName": "Redesign Website", "duration": "3 months" },
{
"projectId": "P002",
"projectName": "Develop Mobile App",
"duration": "6 months"
}
]
},
{
"employeeId": "E002",
"name": {
"first": "Jane",
"last": "Smith"
},
"contact": {
"email": "jane.smith@example.com",
"phones": [
"555-9876"
]
},
"address": {
"street": "456 Oak St",
"city": "Riverside",
"state": "CA",
"zipcode": "92501"
},
"department": {
"deptId": "D002",
"deptName": "Marketing"
},
"projects": [
{
"projectId": "P003",
"projectName": "Product Launch",
"duration": "2 months"
}
]
}
]

PySpark PartitionBy - Example

Read a CSV file and group by Year, for each year write the resulting data in the partition.

df.spark.read.format("csv").option("inferschema", True).option("header", True).option("sep", ",").load("/FileStore/tables/baby_name/input/")

display(df)

df.groupBy("Year").count().show(truncate=False)

df.write.option("header", True).partitionBy("Year").mode("overwrite").csv("/FileStore/tables/flower_name/output/")

Read a CSV file and group by Year, Color and write it by Year, for each year write the resulting data in the partition.

df.write.option("header", True).partitionBy("Year", "Color").mode("overwrite").csv("/FileStore/tables/flower_name/output/")

Read a CSV file, partition it based on the number of records, in this case 4000 records per partition, for each year write the resulting data in the partition.artition

df.write.option("header", True).option("maxRecordsPerFile", 4200).partitionBy("Year").mode("overwrite").csv("/FileStore/tables/flower_name/output/")

dbutils.fs.rm("/FileStore/tables/flower_name/output/", True)

dbutils.fs.mkdir("/FileStore/tables/flower_name/output/")

Below is an example of how you might partition a dataset of employee records by a year column using PySpark.

from pyspark.sql import SparkSession
from pyspark.sql.functions import year, to_date, col
Initialize the Spark session

spark = SparkSession.builder \
.appName("EmployeeRecordsPartitioning") \
.getOrCreate()

Read the employee records from a CSV file. Adjust the file path and options as needed.

df = spark.read.csv("employee_records.csv", header=True, inferSchema=True)

Optional: Convert the 'hire_date' column from string to date format.
(Assumes hire_date is stored in "yyyy-MM-dd" format)

df = df.withColumn("hire_date", to_date(col("hire_date"), "yyyy-MM-dd"))

Extract the year from the 'hire_date' column. If your dataset already has a year column, this step isn’t necessary.
df = df.withColumn("year", year(col("hire_date")))
Display a few rows to verify the new 'year' column.
df.show()

Write the DataFrame partitioning the records into separate directories by the 'year' value.
The resulting partitioned data is stored in Parquet format at the specified output path.
output_path = "/path/to/output/employee_partitioned"
df.write.mode("overwrite").partitionBy("year").parquet(output_path)
Stop the Spark session
spark.stop()

Performing Incremental Data Loads

Performing Incremental Data Loads
When your data source continuously generates new or updated records, you don’t want to reload the entire dataset each time. Instead, you can load only the changes (i.e., incremental data) and merge them into your Delta table. Delta Lake provides the powerful MERGE API to do this.

Example: Upsert New Records Using Delta Lake’s MERGE API

Suppose you have a Delta table stored at /mnt/delta/my_table and you receive a new batch of records as a DataFrame called new_data_df. You can use the following code to merge (upsert) the incremental changes:

from delta.tables import DeltaTable
from pyspark.sql.functions import current_timestamp
# Example incremental data (new or updated rows)

new_data = [
(1, "Alice", 70000.0),
(3, "Charlie", 80000.0) # new record with id=3
]

columns = ["id", "name", "salary"]
new_data_df = spark.createDataFrame(new_data, columns)

Reference the existing Delta table
deltaTable = DeltaTable.forPath(spark, "/mnt/delta/my_table")

Perform MERGE to upsert new data

deltaTable.alias("t").merge(
new_data_df.alias("s"),
"t.id = s.id" # join condition: match records on id
).whenMatchedUpdate(
set={
"name": "s.name",
"salary": "s.salary",
"last_updated": "current_timestamp()"
}
).whenNotMatchedInsert(
values={
"id": "s.id",
"name": "s.name",
"salary": "s.salary",
"last_updated": "current_timestamp()"
}
).execute()

Explanation:

Merge Operation: The code matches incoming rows (s) to existing rows (t) based on the id column.
When Matched: If the record exists, it updates certain columns (and records the update time).
When Not Matched: If no match is found, it inserts the new record into the table.
This incremental load avoids reprocessing your entire dataset every time new data arrives, making your processes efficient—ideal for real-time or near-real-time analytics .

Commonly used DataFrame functions

Data Manipulation Functions

1. select(): Selects a subset of columns from the DataFrame.
2. filter(): Filters the DataFrame based on a condition.
3. where(): Similar to filter(), but allows for more complex conditions.
4. groupBy(): Groups the DataFrame by one or more columns.
5. agg(): Performs aggregation operations on the grouped DataFrame.
6. join(): Joins two DataFrames based on a common column.
7. union(): Combines two DataFrames into a single DataFrame.
8. intersect(): Returns the intersection of two DataFrames.
9. exceptAll(): Returns the difference between two DataFrames.

Data Transformation Functions

1. withColumn(): Adds a new column to the DataFrame.
2. withColumnRenamed(): Renames an existing column in the DataFrame.
3. drop(): Drops one or more columns from the DataFrame.
4. cast(): Casts a column to a different data type.
5. orderBy(): Sorts the DataFrame by one or more columns.
6. sort(): Similar to orderBy(), but allows for more complex sorting.
7. repartition(): Repartitions the DataFrame into a specified number of partitions.

Data Analysis Functions

1. count(): Returns the number of rows in the DataFrame.
2. sum(): Returns the sum of a column in the DataFrame.
3. avg(): Returns the average of a column in the DataFrame.
4. max(): Returns the maximum value of a column in the DataFrame.
5. min(): Returns the minimum value of a column in the DataFrame.
6. groupBy().pivot(): Pivots the DataFrame by a column and performs aggregation.
7. corr(): Returns the correlation between two columns in the DataFrame.


Data Inspection Functions

1. show(): Displays the first few rows of the DataFrame.
2. printSchema(): Prints the schema of the DataFrame.
3. dtypes: Returns the data types of the columns in the DataFrame.
4. columns: Returns the column names of the DataFrame.
5. head(): Returns the first few rows of the DataFrame.

Why user defined function should be wrapped using UDF()

In PySpark, the udf function is used to wrap a user-defined function (UDF) so that it can be used with PySpark DataFrames. Here are some reasons why you should use the udf function:
1. Type Safety: When you use the udf function, you need to specify the return type of the UDF. This helps catch type-related errors at runtime.
2. Serialization: PySpark needs to serialize the UDF and send it to the executors. The udf function takes care of serializing the UDF.
3. Registration: The udf function registers the UDF with PySpark, making it available for use with DataFrames.
4. Optimization: PySpark can optimize the execution of the UDF, such as reusing the UDF across multiple rows.
5. Integration with PySpark API: The udf function allows you to integrate your UDF with the PySpark API, making it easier to use with DataFrames. from pyspark.sql.functions import udf
from pyspark.sql.types import StringType


# Define the UDF
def to_uppercase(s):
return s.upper()

# Wrap the UDF with the udf function
udf_to_uppercase = udf(to_uppercase, StringType())
# Use the UDF with a DataFrame
df = spark.createDataFrame([("John",), ("Mary",)], ["Name"])
df_uppercase = df.withColumn("Name_Uppercase", udf_to_uppercase(df["Name"]))
# Print the result
df_uppercase.show()

Read csv file and plot a graph

from pyspark.sql import SparkSession
import matplotlib.pyplot as plt
import os
import time
import glob
import argparse

# Create SparkSession
spark = SparkSession.builder.appName("Employee Graph").getOrCreate()
def read_csv_file(file_path):

# Read CSV file into DataFrame
df = spark.read.csv(file_path, header=True, inferSchema=True)

# Register DataFrame as temporary view
df.createOrReplaceTempView("employees")
# Count employees by department

department_counts = spark.sql("SELECT department, COUNT(*) as count FROM employees GROUP BY department").toPandas()
return department_counts

def create_bar_chart(department_counts):
plt.bar(department_counts['department'], department_counts['count'])
plt.xlabel('Department')
plt.ylabel('Number of Employees')
plt.title('Employee Distribution by Department')
plt.show()

def create_pie_chart(department_counts):
plt.pie(department_counts['count'], labels=department_counts['department'], autopct='%1.1f%%')
plt.title('Employee Distribution by Department')
plt.show()

def add_employee_to_db(df):
# Write DataFrame to SQLite database
df.write.format("jdbc").option("url", "jdbc:sqlite:employees.db").option("driver", "org.sqlite.JDBC").option("dbtable", "employees").save()

def main():
parser = argparse.ArgumentParser(description='Employee Graph')
parser.add_argument('--chart', choices=['bar', 'pie'], help='Type of chart to create')
args = parser.parse_args()

while True:
csv_files = glob.glob('*.csv')
if len(csv_files) > 0:
department_counts = read_csv_file(csv_files[-1])
if args.chart == 'bar':
create_bar_chart(department_counts)
elif args.chart == 'pie':
create_pie_chart(department_counts)

# Add employees to SQLite database
df = spark.read.csv(csv_files[-1], header=True, inferSchema=True)
add_employee_to_db(df)
time.sleep(60)

if __name__ == "__main__":

main()
Note that this code uses the pyspark.sql module to read and manipulate the CSV data, and the matplotlib library to create the charts. The add_employee_to_db function writes the DataFrame to a SQLite database using the jdbc format.

Processing RDD and the use of Parallelize

from pyspark import SparkContext
# Initialize SparkContext
sc = SparkContext("local", "RDD Example")
# Create an RDD from a list of data
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
rdd = sc.parallelize(data)
# Perform some basic operations on the RDD
# 1. Collect: Gather all elements of the RDD
collected_data = rdd.collect()
print("Collected data:", collected_data)
# 2. Map: Apply a function to each element
mapped_rdd = rdd.map(lambda x: x * 2)
print("Mapped data:", mapped_rdd.collect())
# 3. Filter: Filter elements based on a condition
print("Filtered data (even numbers):", filtered_rdd.collect())
# 4. Reduce: Aggregate elements using a function
sum_of_elements = rdd.reduce(lambda x, y: x + y)
print("Sum of elements:", sum_of_elements)
# 5. Count: Count the number of elements in the RDD
count_of_elements = rdd.count()
print("Count of elements:", count_of_elements)
# Stop the SparkContext
sc.stop()

Explanation: We initialize a SparkContext with a local master.

We create an RDD from a list of integers using the parallelize method.
We perform various operations on the RDD:
Collect: Gather all elements of the RDD and print them.
Map: Apply a function to each element (in this case, multiply by 2) and print the result.
Filter: Filter elements based on a condition (even numbers) and print the result.
Reduce: Aggregate elements by summing them and print the result.
Count: Count the number of elements in the RDD and print the result.



Using the parallelize method in PySpark is essential for several reasons:

Creating RDDs: parallelize allows you to create an RDD (Resilient Distributed Dataset) from an existing collection, such as a list or array. RDDs are fundamental data structures in Spark, enabling distributed data processing and fault tolerance.

Parallel Processing: When you use parallelize, the data is automatically distributed across the available computing resources (nodes) in the cluster. This means that operations on the RDD can be executed in parallel, significantly speeding up data processing.

Scalability: By parallelizing data, you can handle large datasets that wouldn't fit into the memory of a single machine. Spark distributes the data across the cluster, allowing you to process massive amounts of data efficiently.

Fault Tolerance: RDDs provide fault tolerance through lineage information. If a node fails during computation, Spark can recompute the lost data using the lineage information. This ensures the reliability of your data processing pipeline.
Ease of Use: The parallelize method simplifies the process of creating RDDs. You can quickly convert existing collections into RDDs and start applying transformations and actions using Spark's powerful API.
Here's a quick example to illustrate the use of parallelize:
from pyspark import SparkContext
# Initialize SparkContext
sc = SparkContext("local", "Parallelize Example")
# Create an RDD from a list of data using parallelize
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
rdd = sc.parallelize(data)
# Perform a simple transformation (map) and action (collect)
result = rdd.map(lambda x: x * 2).collect()
# Print the result
print("Result:", result)
# Stop the SparkContext
sc.stop()
In this example, parallelize creates an RDD from a list of integers, and the data is distributed across the cluster for parallel processing. We then apply a simple map transformation to double each element and collect the results

PySpark - Commonly used functions

Data Manipulation

1. withColumn(): Add a new column to a DataFrame.
2. withColumnRenamed(): Rename an existing column.
3. drop(): Drop one or more columns from a DataFrame.
4. cast(): Cast a column to a different data type.

Data Analysis

1. count(): Count the number of rows in a DataFrame.
2. sum(): Calculate the sum of a column.
3. avg(): Calculate the average of a column.
4. max(): Find the maximum value in a column.
5. min(): Find the minimum value in a column.

Data Transformation

1. explode(): Transform an array column into separate rows.
2. flatten(): Flatten a nested struct column.
3. split(): Split a string column into an array.

Dataframe Operations

1. distinct(): Return a DataFrame with unique rows.
2. intersect(): Return a DataFrame with rows common to two DataFrames
3. exceptAll(): Return a DataFrame with rows in the first DataFrame but not in the second.
4. repartition(): Repartition a DataFrame to increase or decrease the number of partitions.
5. coalesce(): Coalesce a DataFrame to reduce the number of partitions.

Data Manipulation

1. orderBy(): Order a DataFrame by one or more columns.
2. sort(): Sort a DataFrame by one or more columns.
3. limit(): Limit the number of rows in a DataFrame.
4. sample(): Return a sampled subset of a DataFrame.
5. randomSplit(): Split a DataFrame into multiple DataFrames randomly.

Data Analysis

1. corr(): Calculate the correlation between two columns.
2. cov(): Calculate the covariance between two columns.
3. skewness(): Calculate the skewness of a column.
4. kurtosis(): Calculate the kurtosis of a column.
5. approxQuantile(): Calculate an approximate quantile of a column.

Data Transformation

1. udf(): Create a user-defined function (UDF) to transform data.
2. apply(): Apply a UDF to a column.
3. transform(): Transform a DataFrame using a UDF.
4. map(): Map a DataFrame to a new DataFrame using a UDF.

String Functions

1. concat(): Concatenate two or more string columns.
2. length(): Calculate the length of a string column.
3. lower(): Convert a string column to lowercase.
4. upper(): Convert a string column to uppercase.
5. trim(): Trim whitespace from a string column.

Date and Time Functions

1. current_date(): Return the current date.
2. current_timestamp(): Return the current timestamp.
3. date_format(): Format a date column.
4. hour(): Extract the hour from a timestamp column.
5. dayofweek(): Extract the day of the week from a date column.

PySpark - Union (combine two data frames)

from pyspark.sql import SparkSession
# Create a Spark session
spark = SparkSession.builder.appName("example").getOrCreate()
# Sample data for DataFrame 1
data1 = [("Alice", 25, "New York"),
("Bob", 30, "Los Angeles"),
("Charlie", 35, "Chicago")]

# Sample data for DataFrame 2
data2 = [("David", 40, "San Francisco"),
("Eve", 45, "Miami"),
("Frank", 50, "Seattle")]


# Create DataFrames
columns = ["Name", "Age", "City"]
df1 = spark.createDataFrame(data1, columns)
df2 = spark.createDataFrame(data2, columns)
# Show the DataFrames
df1.show()
df2.show()
# Perform a union operation
union_df = df1.union(df2)
# Show the result
union_df.show()

PySpark - Inner Join

from pyspark.sql import SparkSession # Create a Spark session
spark = SparkSession.builder.appName("example").getOrCreate()
# Sample data for DataFrame 1
data1 = [("Alice", 25, "New York"), ("Bob", 30, "Los Angeles"), ("Charlie", 35, "Chicago")]
# Sample data for DataFrame 2
data2 = [("Alice", "F"), ("Bob", "M"), ("Charlie", "M")]
# Create DataFrames

columns1 = ["Name", "Age", "City"]
columns2 = ["Name", "Gender"]
df1 = spark.createDataFrame(data1, columns1)
df2 = spark.createDataFrame(data2, columns2)

# Show the DataFrames
df1.show()
df2.show()
# Perform a join operation
joined_df = df1.join(df2, on="Name", how="inner")
# Show the result
joined_df.show()

PySpark Functions - filter, groupBy, agg

from pyspark.sql import SparkSession
# Create a Spark session
spark = SparkSession.builder.appName("example").getOrCreate()
# Sample data
data = [("Alice", 25, "New York"),
("Bob", 30, "Los Angeles"),
("Charlie", 35, "Chicago")]

# Create DataFrame
columns = ["Name", "Age", "City"]
df = spark.createDataFrame(data, columns)
# Show the DataFrame
df.show()
# Filter the DataFrame

filtered_data = df.where(df.Age > 30)
filtered_data.show()

# Group by 'City' and calculate the average 'Age'
grouped_data = df.groupBy("City").agg(avg("Age").alias("Average_Age"))
# Show the result
grouped_data.show()

# Use the agg() function to calculate the average and maximum age for each city
aggregated_data = df.groupBy("City").agg(avg("Age").alias("Average_Age"), max("Age").alias("Max_Age"))
# Show the result
aggregated_data.show()

PySpark - Filter

from pyspark.sql import SparkSession
# Create a Spark session
spark = SparkSession.builder.appName("example").getOrCreate()
# Sample data
data = [("Alice", 25, "New York"),
("Bob", 30, "Los Angeles"),
("Charlie", 35, "Chicago")]

# Create DataFrame
columns = ["Name", "Age", "City"]
df = spark.createDataFrame(data, columns)
# Show the DataFrame
df.show()
# Filter the DataFrame
filtered_data = df.filter(df.Age > 30)
# Show the result
filtered_data.show()

PySpark - Select

from pyspark.sql import SparkSession
# Create a Spark session
spark = SparkSession.builder.appName("example").getOrCreate()
# Sample data
data = [("Alice", 25, "New York"),
("Bob", 30, "Los Angeles"),
("Charlie", 35, "Chicago")]

# Create DataFrame
columns = ["Name", "Age", "City"]
df = spark.createDataFrame(data, columns)
# Select specific columns
selected_columns = df.select("Name", "City")
# Show the result
selected_columns.show()
print(type(selected_columns))

In this example, a Spark session is created, and a DataFrame df is created with the columns 'Name', 'Age', and 'City'. The select method is used to create a new DataFrame selected_columns that only includes the 'Name' and 'City' columns.

How to use Split function in PySpark

from pyspark.sql import SparkSession from pyspark.sql.functions import split

# Initialize a Spark session
spark = SparkSession.builder \ .appName("Split Column") \ .getOrCreate()

# Sample data

data = [
(1, "John Doe"),
(2, "Jane Smith"),
(3, "Alice Johnson")
]

# Create DataFrame from sample data
df = spark.createDataFrame(data, ["id", "full_name"])
# Split the 'full_name' column into 'first_name' and 'last_name'

df_split = df.withColumn("first_name", split(df["full_name"], " ").getItem(0)) \ .withColumn("last_name", split(df["full_name"], " ").getItem(1))

# Show the resulting DataFrame
df_split.show()

# Stop the Spark session

spar
k.stop()


In t his example:
We initialize a Spark session.

We create a DataFrame from sample data with an id column and a full_name column.

We use the split function to split the full_name column into two new colu
mns: first_name and last_name.
We display the resulting DataFrame.

The split function splits the full_name column into an array of s
trings based on the delimiter (a space in this case), and then we use getItem(0) and getItem(1) to extract the first and last names, respectively.

Flatten a nested DataFrame

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, explode

# Initialize a Spark session
spark = SparkSession.builder \ .appName("Flatten DataFrame") \ .getOrCreate()

# Sample nested JSON data

data = [
{
"id": 1,
"name": "John",
"subjects": [{"subject": "Math", "score": 90}, {"subject": "English", "score": 85}]
},
{
"id": 2,
"name": "Jane",
"subjects": [{"subject": "Math", "score": 95}, {"subject": "Science", "score": 80}]
}
]

# Create DataFrame from sample data

df = spark.read.json(spark.sparkContext.parallelize(data))

# Flatten the DataFrame

flattened_df = df.withColumn("subject", explode(col("subjects"))).select("id", "name", col("subject.subject"), col("subject.score"))
# Show the flattened DataFrame
flattened_df.show()
# Stop the Spark session
spark.stop()

In this example:

We initialize a Spark session.
We create a DataFrame from nested JSON data.
We use the explode function to flatten the nested subjects array into individual rows.
We select the flattened columns and display the result.

----------------------------------------------------

The code snippet spark.read.json(spark.sparkContext.parallelize(data)) does the following:

Parallelizes Data: The spark.sparkContext.parallelize(data) part converts the data (which is a list of dictionaries in this case) into an RDD (Resilient Distributed Dataset), which is a fundamental data structure of Spark. This process spreads the data across multiple nodes in the Spark cluster, allowing parallel processing.

Reads JSON Data: The spark.read.json() part reads the parallelized RDD as a JSON dataset and converts it into a DataFrame. This step interprets the structure of the JSON data, defining the schema (the structure of columns) and their data types.

Combining these two parts, the code creates a DataFrame from the JSON data stored in the data variable, enabling you to perform various transformations and actions using Spark DataFrame API.

----------------------------------------------------

If you dont read the parallelized data you need to read from a data file and change the code as follows:

import json
# Write the data to a JSON file
with open('data.json', 'w') as f: json.dump(data, f)
# Read the JSON file into a DataFrame
df = spark.read.json('data.json')
# Show the DataFrame
df.show()

PySpark - Commonly used functions

Dataframe Operations

1. select(): Select specific columns from a DataFrame.
2. filter(): Filter rows based on conditions.
3. where(): Similar to filter(), but uses SQL-like syntax.
4. groupBy(): Group rows by one or more columns.
5. agg(): Perform aggregation operations (e.g., sum, count, avg).
6. join(): Join two DataFrames based on a common column.
7. union(): Combine two DataFrames into a single DataFrame.

Data Manipulation

1. withColumn(): Add a new column to a DataFrame.
2. withColumnRenamed(): Rename an existing column.
3. drop(): Drop one or more columns from a DataFrame.
4. cast(): Cast a column to a different data type.

Data Analysis

1. count(): Count the number of rows in a DataFrame.
2. sum(): Calculate the sum of a column.
3. avg(): Calculate the average of a column.
4. max(): Find the maximum value in a column.
5. min(): Find the minimum value in a column.

Data Transformation

1. explode(): Transform an array column into separate rows.
2. flatten(): Flatten a nested struct column.
3. split(): Split a string column into an array.

Collect - PySpark

Here's a sample of using the collect method in PySpark:
data = [("John", 25), ("Mary", 31), ("David", 42)]
df = spark.createDataFrame(data, ["Name", "Age"])
result = df.collect()
print(result)

Output:

[Row(Name='John', Age=25), Row(Name='Mary', Age=31), Row(Name='David', Age=42)]
The collect method returns all the rows in the DataFrame as a list of Row objects. Note that this can be memory-intensive for large DataFrames.

To display the age only from the first row, you can use the following code:

data = [("John", 25), ("Mary", 31), ("David", 42)]
df = spark.createDataFrame(data, ["Name", "Age"])

first_row = df.first()
age = first_row.Age
print(age)
Output:
25

Alternatively, you can use:
print(df.first().Age)
Both methods will display the age from the first row, which is 25.

In PySpark, collect() returns a list of Row objects.
collect[0] refers to the first Row object in the list.
collect[0][0] refers to the first element (or column value) within that first Row object.
So, collect[0][0] essentially gives you the value of the first column in the first row of the DataFrame.
Here's an example:

data = [("John", 25), ("Mary", 31), ("David", 42)]
df = spark.createDataFrame(data, ["Name", "Age"])
result = df.collect()
print(result[0][0])
# Output: John

Explode - PySpark

In PySpark, the explode function is used to transform each element of a collection-like column (e.g., array or map) into a separate row.

Suppose we have a DataFrame df with a column fruits that contains an array of fruit names:

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
spark = SparkSession.builder.appName("Explode Example").getOrCreate()

data = [ ("John", ["Apple", "Banana", "Cherry"]),
("Mary", ["Orange", "Grapes", "Peach"]),
("David", ["Pear", "Watermelon", "Strawberry"])
]

df = spark.createDataFrame(data, ["name", "fruits"])
df.show()

Display the contents of "fruits" only

df.select("fruits").show(truncate=False)

- John: [Apple, Banana, Cherry]
- Mary: [Orange, Grapes, Peach]
- David: [Pear, Watermelon, Strawberry]

Output:
+-----+--------------------+
| name| fruits|
+-----+--------------------+
| John|[Apple, Banana, ...|
| Mary|[Orange, Grapes, ...|
|David|[Pear, Watermelon...|
+-----+--------------------+

Now, let's use the explode function to transform each element of the fruits array into a separate row:

df_exploded = df.withColumn("fruit", explode("fruits"))
df_exploded.show()

Here's what's happening:

1. df.withColumn(): This method adds a new column to the existing DataFrame df.
2. "fruit": This is the name of the new column being added.
3. explode("fruits"): This is the transformation being applied to create the new column. Output:

+-----+--------------------+------+
| name| fruits| fruit|
+-----+--------------------+------+
| John|[Apple, Banana, ...| Apple|
| John|[Apple, Banana, ...|Banana|
| John|[Apple, Banana, ...|Cherry|
| Mary|[Orange, Grapes, ...|Orange|
| Mary|[Orange, Grapes, ...|Grapes|
| Mary|[Orange, Grapes, ...| Peach|
|David|[Pear, Watermelon...| Pear|
|David|[Pear, Watermelon...|Watermelon|
|David|[Pear, Watermelon...|Strawberry|
+-----+--------------------+------+


As you can see, the explode function has transformed each element of the fruits array into a separate row, with the corresponding name value.

To select only the "name" and "fruit" columns from the df_exploded DataFrame, you can use the select method:

df_name_fruit = df_exploded.select("name", "fruit")
df_name_fruit.show()
Output:
+-----+------+
| name| fruit|
+-----+------+
| John| Apple|
| John|Banana|
| John|Cherry|
| Mary|Orange|
| Mary|Grapes|
| Mary| Peach|
|David| Pear|
|David|Watermelon|
|David|Strawberry|
+-----+------+


By using select, you're creating a new DataFrame (df_name_fruit) that contains only the specified columns.