PySpark Basics

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("PySparkBasics") \
    .master("local[*]") \
    .config("spark.sql.shuffle.partitions", "8") \
    .config("spark.driver.memory", "4g") \
    .getOrCreate()

# Access the underlying SparkContext
sc = spark.sparkContext
print(f"Spark version: {spark.version}")
print(f"Default parallelism: {sc.defaultParallelism}")

# Read from CSV
df = spark.read.csv("customers.csv", header=True, inferSchema=True)

# Read from Parquet (preferred format for ML)
df = spark.read.parquet("features.parquet")

# Read from JSON
df = spark.read.json("events.json")

# Explore the data
df.printSchema()          # Show column names and types
df.show(10)               # Show first 10 rows
df.describe().show()      # Summary statistics
df.count()                # Number of rows
df.columns                # List of column names
df.dtypes                 # Column name-type pairs

from pyspark.sql import functions as F

# === Transformations (lazy - build the plan) ===
filtered = df.filter(F.col("age") > 25)
selected = filtered.select("name", "age", "income")
with_ratio = selected.withColumn("income_per_year", F.col("income") / F.col("age"))
sorted_df = with_ratio.orderBy(F.col("income_per_year").desc())

# === Actions (eager - trigger execution) ===
sorted_df.show(10)           # Display rows
sorted_df.count()            # Count rows
sorted_df.collect()          # Return all rows as list
sorted_df.take(5)            # Return first 5 rows
sorted_df.toPandas()         # Convert to pandas DataFrame

from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType

# Handle missing values
df_clean = df.dropna(subset=["label", "feature1"])
df_filled = df.fillna({"feature1": 0.0, "feature2": "unknown"})

# Type casting
df = df.withColumn("price", F.col("price").cast(DoubleType()))

# Group and aggregate
stats = df.groupBy("category").agg(
    F.count("*").alias("count"),
    F.mean("price").alias("avg_price"),
    F.stddev("price").alias("std_price")
)

# Join datasets
train_df = features_df.join(labels_df, on="user_id", how="inner")

# Split data for ML
train, test = df.randomSplit([0.8, 0.2], seed=42)

# Register a DataFrame as a temporary view
df.createOrReplaceTempView("customers")

# Run SQL queries
result = spark.sql("""
    SELECT category,
           COUNT(*) as num_customers,
           AVG(income) as avg_income,
           PERCENTILE_APPROX(income, 0.5) as median_income
    FROM customers
    WHERE age BETWEEN 18 AND 65
    GROUP BY category
    HAVING COUNT(*) > 100
    ORDER BY avg_income DESC
""")
result.show()