DataFrame Operations in PySpark

Spark tutorial · PySpark.in

DataFrame Operations in PySpark

PySpark provides a rich set of operations to transform and manipulate DataFrames — just like SQL queries but optimized for distributed processing.

Transformations (commonly used in ETL & Data Engineering)

• Selecting columns
• Filtering rows
• Sorting
• Dropping columns
• Removing duplicates
• Renaming columns
• Adding new columns
• Type casting
• Working with arrays (split, explode, struct)
  
  

Note: All transformations are lazy — they only define the plan. Execution happens when we call an action like .show(), .count().

Selecting Columns

Selecting columns means retrieving only the required fields from a DataFrame using functions like select() or col().It improves performance by working only with necessary data (similar to SQL SELECT).

```

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df.select("name").show() # single column

df.select("name", "age").show() # multiple columns

df.select(col("age") + 5).show() # expression

df.select(upper(col("country")).alias("UP")).show()

```

Filtering Rows (WHERE clause)

Filtering allows you to keep only those records that match a condition using filter() or where().
Equivalent to SQL WHERE clause — commonly used for data cleaning and business rules application.

```

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df.filter(df.age > 25).show()

df.where("country = 'India'").show()

df.filter(col("name").startswith("A")).show()

```

Avoid using too many .filter() separately — chain them instead.

3 Sorting Data (orderBy / sort)

Sorting arranges rows in ascending or descending order based on one or more columns.
Useful for reports, ranking logic, and readable outputs — but note: sorting can trigger an expensive shuffle.

```

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df.orderBy("age").show() # ascending

df.orderBy(col("age").desc()).show() # descending

df.sort(col("name")).show()

```

Sorting triggers shuffle → high cost in large data.

4  Dropping Columns

Removes unwanted columns from a DataFrame using .drop().
This helps reduce memory usage, avoid duplicate fields after joins, and keep schemas clean.

```

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df.drop("country").show()

df.drop(*["age", "country"]).show()

```

Particularly useful after joins to remove duplicate fields.

5 Removing Duplicate Records

The dropDuplicates() method removes duplicate rows across selected columns or entire DataFrame.Essential for data quality, CDC, and building unique master datasets.

```

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df.dropDuplicates().show() # full row duplicate removal

df.dropDuplicates(["name"]).show() # unique by column(s)

```

6 Renaming Columns

withColumnRenamed() is used to change column names — improving clarity, standardizing naming conventions, and matching schema.

```

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df.withColumnRenamed("country", "home_country").show()

```

7 Adding New Columns (withColumn)

withColumn() helps create new fields derived from existing columns — such as transformations, KPI calculations, metadata fields (load date), or encryption output

```

from pyspark.sql.functions import current_date

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df.withColumn("age_plus_10", col("age") + 10).show()

df.withColumn("ingest_date", current_date()).show()

```

Also used for masking, enrichment, status flags, etc.

8 Type Casting (cast)

cast() converts a column from one data type to another — e.g., string to integer or timestamp.
Crucial when reading CSVs or raw data where types are incorrectly inferred.

```

from pyspark.sql.functions import current_date

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df.withColumn("age", col("age").cast("int")).printSchema()

```

9 Arrays & Complex Column Operations

PySpark supports complex data operations:

a️ Split Strings into Array

In PySpark, the split() function is used to convert a string column into an array by breaking the text using a specified delimiter (like a comma, space, or pipe).

```

from pyspark.sql.functions import split

from pyspark.sql.functions import current_date

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df2 = df.withColumn("name_chars", split(col("name"), ""))

df2.show(truncate=False)

```

b️ Explode Array into Multiple Rows

In PySpark, the explode() function is used to transform each element of an array into its own separate row — while repeating all other column values. It is one of the most powerful functions when dealing with arrays, lists, multi-value fields, JSON records, and denormalized data.

```

from pyspark.sql.functions import explode

from pyspark.sql.functions import current_date

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df2.select("name", explode("name_chars")).show()

```

Create Struct (Nested Object)

A Struct in PySpark is a nested data type that allows you to group multiple related fields into a single column — just like a nested JSON object. It is created using StructType and StructField.

```

from pyspark.sql.functions import struct

from pyspark.sql.functions import current_date

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, upper

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [("Alice", 25, "India"), ("Bob", 30, "USA")]

df = spark.createDataFrame(data, ["name", "age", "country"])

df3 = df.withColumn("details", struct("age", "country"))

df3.show(truncate=False)

```

Real-World ETL Example (All Operations Together)

```

from pyspark.sql import SparkSession

from pyspark.sql.functions import col, split, explode, current_timestamp

spark = SparkSession.builder.appName("Ops").getOrCreate()

data = [

("John Doe", "john@example.com", "India,USA"),

("Megha Kapoor", "megha@abc.com", "UK,India")

]

df = spark.createDataFrame(data, ["name", "email", "countries"])

df_clean = (

df

.withColumn("email_domain", split(col("email"), "@")[1]) # add new column

.withColumn("country", explode(split(col("countries"), ","))) # array explode

.drop("countries") # drop column

.filter(col("country") != "USA") # filter

.orderBy("name") # sort

.withColumnRenamed("name", "full_name") # rename

.withColumn("load_time", current_timestamp()) # metadata column

)

df_clean.show(truncate=False)

```

Transformation Table:

More Spark tutorials

• about pyspark
• text diagram
• Apache Spark Runtime Architecture
• Introduction to RDD
• Actions vs Transformations
• Lazy Evaluation in PySpark

All tutorials · Try the free PySpark compiler · Practice challenges