PySpark Operations on DataFrames

---

PySpark DataFrame operations can be broadly classified into two categories: Transformations and Actions.

---

Transformations

Transformations are operations that are applied on a DataFrame to produce a new DataFrame. Transformations are lazy, meaning they are not executed immediately. Instead, they are recorded as a lineage of transformations to be applied when an action is called.

Here are some common transformations:

select()

• Description: Selects a subset of columns.
• Syntax: df.select("col1", "col2", ...)
• Example: df.select("name", "age").show()

df_pyspark.select([‘Vars’,’Month_period’]).show()
df_pyspark.select(‘Vars’,’Month_period’).show()
from pyspark.sql.functions import col
df_pyspark.select(col(“Month_Period”)).show()
df_pyspark[“Month_Period”] are all these same

withColumn()

• Description: Adds or replaces a column.
• Syntax: df.withColumn("new_col", df["existing_col"] + 1)
• Example:df.withColumn("new_age", df["age"] + 1).show()

use cases for the withColumn() method in PySpark:

1. Conditional Column Creation

from pyspark.sql.functions import when, col

df = df.withColumn("is_adult", when(col("age") >= 18, "Yes").otherwise("No"))

2. Column Transformation using UDFs

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

def convert_to_uppercase(name):
    return name.upper()

udf_convert_to_uppercase = udf(convert_to_uppercase, StringType())

df = df.withColumn("uppercase_name", udf_convert_to_uppercase(df["name"]))

3. Column Calculation using Multiple Columns

from pyspark.sql.functions import col

df = df.withColumn("total_amount", col("quantity") * col("price"))

4. Renaming and Dropping Columns

df = df.withColumnRenamed("old_name", "new_name")
df = df.drop("column_to_drop")

5. Creating a Column with a List or Array

from pyspark.sql.functions import array, lit

df = df.withColumn("colors", array(lit("red"), lit("green"), lit("blue")))

6. Creating a Column with a Struct

from pyspark.sql.functions import struct, lit

df = df.withColumn("address", struct(lit("123 Main St"), lit("Anytown"), lit("USA")))

7. Using Window Functions

from pyspark.sql.functions import row_number
from pyspark.sql.window import Window

window = Window.orderBy("date")

df = df.withColumn("row_num", row_number().over(window))

8. Creating a Column with a JSON Object

from pyspark.sql.functions import to_json, struct

df = df.withColumn("json_data", to_json(struct(df["name"], df["age"])))

9. Using Aggregate Functions

from pyspark.sql.functions import sum, avg

df = df.groupBy("group").agg(sum("value").alias("sum_value"), avg("value").alias("avg_value"))
df = df.withColumn("sum_value_squared", df["sum_value"] ** 2)

10. Using Pattern Matching with when and `otherwise

from pyspark.sql.functions import when, col

df = df.withColumn("category",
                   when(col("value") < 10, "low").
                   when(col("value") < 20, "medium").
                   otherwise("high"))

drop()

• Description: Drops specified columns.
• Syntax: df.drop("col1", "col2")
• Example:df.drop("age").show()

distinct()

• Description: Returns distinct rows.
• Syntax: df.distinct()
• Example:df.distinct().show()

filter() or where()

• Description: Filters rows based on a condition.
• Syntax: df.filter(df["age"] > 30)
• Example:df.filter(df[“age”] > 30).show()

 

groupBy()

• Description: Groups rows by specified columns.
• Syntax: df.groupBy("department").agg({"salary": "mean"})
• Example: df.groupBy("department").mean("salary").show()

orderBy() or sort()

• Description: Sorts rows by specified columns.
• Syntax: df.orderBy(df["age"].desc())
• Example:df.orderBy(df["age"].desc()).show()

join()

• Description: Joins two DataFrames.
• Syntax: df1.join(df2, df1["key"] == df2["key"], "inner")
• Example:df1.join(df2, df1["id"] == df2["id"], "inner").show()

union()

• Description: Unions two DataFrames.
• Syntax: df1.union(df2)
• Example:df1.union(df2).show()

sample()

• Description: Returns a sampled subset of rows.
• Syntax: df.sample(withReplacement=False, fraction=0.1)
• Example:df.sample(withReplacement=False, fraction=0.1).show()

---

Actions

Actions trigger the execution of the transformations recorded in the DAG and return results.

Here are some common actions:

collect()

• Description: Returns all rows as a list of Row objects.
• Syntax: df.collect()
• Example:rows = df.collect() for row in rows: print(row)

first()

• Description: Returns the first row.
• Syntax: df.first()
• Example:print(df.first())

take()

• Description: Returns the first n rows.
• Syntax: df.take(n)
• Example:print(df.take(5))

describe()

• Description: Computes basic statistics.
• Syntax: df.describe()
• Example:df.describe().show()

cache()

• Description: Caches the DataFrame in memory.
• Syntax: df.cache()
• Example:df.cache()

persist()

• Description: Persists the DF with specified storage level.
• Syntax: df.persist(storageLevel)
• Example:from pyspark import StorageLevel df.persist(StorageLevel.MEMORY_AND_DISK)

write()

• Description: Writes the DataFrame to a specified format.
• Syntax: df.write.format("parquet").save("path")
• Example:df.write.format(“parquet”).save(“path/to/parquet”)

foreach()

• Description: Applies a function to each row.
• Syntax: df.foreach(f)
• Example:def print_row(row): print(row) df.foreach(print_row)

---

---

Head to Next

Pages: 1 2 3 4 5 6 7 8 9 10 11 12