1. PySpark Cheat Sheet

    ┌─────────────────────────────────────────────────┐
    │              Driver Program                      │
    │  ┌──────────────────────────────────────────┐   │
    │  │        SparkContext/SparkSession         │   │
    │  └──────────────┬───────────────────────────┘   │
    └─────────────────┼───────────────────────────────┘
                      │
           ┌──────────┴──────────┐
           │   Cluster Manager   │
           │  (YARN/Mesos/K8s)   │
           └──────────┬──────────┘
                      │
        ┌─────────────┼─────────────┐
        ↓             ↓             ↓
    ┌─────────┐  ┌─────────┐  ┌─────────┐
    │ Worker  │  │ Worker  │  │ Worker  │
    │  Node   │  │  Node   │  │  Node   │
    │┌───────┐│  │┌───────┐│  │┌───────┐│
    ││Executor││ ││Executor││ ││Executor││
    ││ Cache  ││  ││ Cache  ││  ││ Cache  ││
    ││ Tasks  ││  ││ Tasks  ││  ││ Tasks  ││
    │└───────┘│  │└───────┘│  │└───────┘│
    └─────────┘  └─────────┘  └─────────┘

# Basic installation
pip install pyspark

# With specific version
pip install pyspark==3.5.0

# With additional dependencies
pip install pyspark[sql,ml,streaming]

# Create and activate virtual environment
python -m venv venv
source venv/bin/activate  # On Linux/macOS
venv\Scripts\activate     # On Windows

# Install PySpark
pip install pyspark

from pyspark.sql import SparkSession

# Basic session
spark = SparkSession.builder \
    .appName("MyPySparkApp") \
    .master("local[*]") \
    .getOrCreate()

# With configurations
spark = SparkSession.builder \
    .appName("MyPySparkApp") \
    .master("local[*]") \
    .config("spark.executor.memory", "4g") \
    .config("spark.driver.memory", "2g") \
    .config("spark.sql.shuffle.partitions", "200") \
    .enableHiveSupport() \
    .getOrCreate()

# Access SparkContext from SparkSession
sc = spark.sparkContext

# Get Spark version
print(spark.version)

from pyspark import SparkContext, SparkConf

# Configuration
conf = SparkConf() \
    .setAppName("MyPySparkApp") \
    .setMaster("local[*]") \
    .set("spark.executor.memory", "4g")

# Create SparkContext
sc = SparkContext(conf=conf)

# Create SparkSession from existing context
from pyspark.sql import SparkSession
spark = SparkSession(sparkContext=sc)

# Stop SparkSession (also stops SparkContext)
spark.stop()

# Stop SparkContext only
sc.stop()

    ┌──────────────┐
    │ Data Sources │
    └──────┬───────┘
           │
    ┌──────┴──────────────────────────────────┐
    │                                          │
    ↓                ↓            ↓            ↓
┌────────┐      ┌──────┐    ┌────────┐   ┌──────────┐
│  Text  │      │ CSV  │    │ JSON   │   │  JDBC    │
│ Files  │      │Files │    │ Files  │   │Database  │
└───┬────┘      └───┬──┘    └───┬────┘   └────┬─────┘
    │               │           │             │
    └───────────────┴───────────┴─────────────┘
                    │
                    ↓
           ┌────────────────┐
           │  Spark Reader  │
           └────────┬───────┘
                    │
                    ↓
           ┌────────────────┐
           │   DataFrame    │
           └────────────────┘

# Using SparkContext (RDD)
lines = sc.textFile("path/to/textfile.txt")
lines = sc.textFile("hdfs://path/to/textfile.txt")  # HDFS
lines = sc.textFile("s3a://bucket/path/file.txt")   # S3

# Multiple files
lines = sc.textFile("path/to/directory/*.txt")

# Using SparkSession (DataFrame)
df = spark.read.text("path/to/textfile.txt")

# Read multiple text files
df = spark.read.text(["file1.txt", "file2.txt"])

# Basic CSV read
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# With detailed options
df = spark.read.csv(
    "path/to/file.csv",
    header=True,
    inferSchema=True,
    sep=",",
    quote='"',
    escape="\\",
    nullValue="NULL",
    dateFormat="yyyy-MM-dd",
    timestampFormat="yyyy-MM-dd HH:mm:ss"
)

# Alternative syntax
df = spark.read \
    .format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .option("mode", "DROPMALFORMED") \
    .load("path/to/file.csv")

# With explicit schema
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("city", StringType(), True)
])

df = spark.read.csv("path/to/file.csv", schema=schema, header=True)

# Read multiple CSV files
df = spark.read.csv("path/to/directory/*.csv", header=True, inferSchema=True)

# Basic JSON read
df = spark.read.json("path/to/file.json")

# With options
df = spark.read \
    .format("json") \
    .option("multiLine", "true") \
    .option("mode", "PERMISSIVE") \
    .load("path/to/file.json")

# Read JSON lines (one JSON object per line)
df = spark.read.json("path/to/jsonlines.jsonl")

# With schema
df = spark.read.json("path/to/file.json", schema=schema)

# Read multiple JSON files
df = spark.read.json("path/to/directory/*.json")

# Basic Parquet read (preserves schema automatically)
df = spark.read.parquet("path/to/file.parquet")

# Read multiple Parquet files
df = spark.read.parquet("path/to/directory/*.parquet")

# Read partitioned data
df = spark.read.parquet("path/to/partitioned_data")

# With merge schema (for schema evolution)
df = spark.read \
    .option("mergeSchema", "true") \
    .parquet("path/to/file.parquet")

# Basic ORC read
df = spark.read.orc("path/to/file.orc")

# With options
df = spark.read \
    .format("orc") \
    .option("mergeSchema", "true") \
    .load("path/to/file.orc")

# Read multiple ORC files
df = spark.read.orc("path/to/directory/*.orc")

# Basic Avro read
df = spark.read.format("avro").load("path/to/file.avro")

# Read multiple Avro files
df = spark.read.format("avro").load("path/to/directory/*.avro")

# Basic JDBC read
df = spark.read.format("jdbc") \
    .option("url", "jdbc:postgresql://localhost:5432/mydatabase") \
    .option("dbtable", "mytable") \
    .option("user", "myuser") \
    .option("password", "mypassword") \
    .option("driver", "org.postgresql.Driver") \
    .load()

# With query
df = spark.read.format("jdbc") \
    .option("url", "jdbc:postgresql://localhost:5432/mydatabase") \
    .option("query", "SELECT * FROM mytable WHERE age > 25") \
    .option("user", "myuser") \
    .option("password", "mypassword") \
    .load()

# With partitioning for parallel reads
df = spark.read.format("jdbc") \
    .option("url", "jdbc:mysql://localhost:3306/mydb") \
    .option("dbtable", "mytable") \
    .option("user", "myuser") \
    .option("password", "mypassword") \
    .option("partitionColumn", "id") \
    .option("lowerBound", "1") \
    .option("upperBound", "100000") \
    .option("numPartitions", "10") \
    .load()

# Common JDBC URLs
# PostgreSQL: jdbc:postgresql://host:5432/database
# MySQL: jdbc:mysql://host:3306/database
# SQL Server: jdbc:sqlserver://host:1433;databaseName=database
# Oracle: jdbc:oracle:thin:@host:1521:database

from delta.tables import DeltaTable

# Read Delta table
df = spark.read.format("delta").load("path/to/delta_table")

# Alternative: Use DeltaTable
deltaTable = DeltaTable.forPath(spark, "path/to/delta_table")
df = deltaTable.toDF()

# Read specific version (time travel)
df = spark.read.format("delta") \
    .option("versionAsOf", 0) \
    .load("path/to/delta_table")

# Read at specific timestamp
df = spark.read.format("delta") \
    .option("timestampAsOf", "2023-01-01 00:00:00") \
    .load("path/to/delta_table")

# Hive table
df = spark.table("database.tablename")
df = spark.sql("SELECT * FROM database.tablename")

# From Pandas DataFrame
import pandas as pd
pandas_df = pd.DataFrame({"name": ["Alice", "Bob"], "age": [25, 30]})
spark_df = spark.createDataFrame(pandas_df)

# From Python collections
data = [("Alice", 25), ("Bob", 30)]
df = spark.createDataFrame(data, ["name", "age"])

# From RDD
rdd = sc.parallelize([("Alice", 25), ("Bob", 30)])
df = spark.createDataFrame(rdd, ["name", "age"])

    ┌──────────────┐
    │ Data Source  │
    └──────┬───────┘
           │
           ↓
    ┌──────────────┐
    │   Load Data  │
    └──────┬───────┘
           │
           ↓
    ┌──────────────┐
    │ Transform    │────┐
    └──────┬───────┘    │
           │            │
           ↓            │
    ┌──────────────┐    │
    │   Action     │    │
    └──────┬───────┘    │
           │            │
           ↓ (Trigger)  │
    ┌──────────────┐    │
    │   Execute    │    │
    │  Lazy Eval   │    │
    └──────┬───────┘    │
           │            │
           ↓            │
    ┌──────────────┐    │
    │   Result     │    │
    └──────────────┘    │
                        │
           More Transforms? ──┘

# With column names
data = [("Alice", 30), ("Bob", 25), ("Charlie", 35)]
df = spark.createDataFrame(data, ["Name", "Age"])

# Inferred schema
data = [("Alice", 30, "Engineer"), ("Bob", 25, "Doctor")]
df = spark.createDataFrame(data)
df.show()

# Basic RDD to DataFrame
data = [("Alice", 30), ("Bob", 25)]
rdd = spark.sparkContext.parallelize(data)
df = spark.createDataFrame(rdd, schema=["Name", "Age"])

# With explicit schema
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("Name", StringType(), nullable=False),
    StructField("Age", IntegerType(), nullable=True)
])
df = spark.createDataFrame(rdd, schema=schema)

data = [
    {"Name": "Alice", "Age": 30, "City": "NYC"},
    {"Name": "Bob", "Age": 25, "City": "LA"}
]
df = spark.createDataFrame(data)

import pandas as pd

# Convert Pandas to Spark DataFrame
pandas_df = pd.DataFrame({
    "Name": ["Alice", "Bob"],
    "Age": [30, 25]
})
spark_df = spark.createDataFrame(pandas_df)

# Convert Spark to Pandas DataFrame
pandas_df = spark_df.toPandas()

from pyspark.sql import Row

# Create Row objects
data = [
    Row(Name="Alice", Age=30),
    Row(Name="Bob", Age=25)
]
df = spark.createDataFrame(data)

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("Name", StringType(), True),
    StructField("Age", IntegerType(), True)
])
df = spark.createDataFrame([], schema)

### DataFrame Operations

Basic inspection:

```python
# Display rows
df.show()                    # Show first 20 rows
df.show(5)                   # Show first 5 rows
df.show(truncate=False)      # Show without truncating
df.show(vertical=True)       # Vertical format

# Schema and structure
df.printSchema()             # Print schema tree
df.schema                    # Get schema object
df.columns                   # Get column names list
df.dtypes                    # Get column types [(name, type)]

# Statistics
df.count()                   # Count rows
df.describe().show()         # Summary statistics
df.summary("count", "mean", "stddev", "min", "max").show()

# Sample data
df.head(5)                   # Return first 5 rows as list
df.first()                   # Return first row
df.take(5)                   # Return first 5 rows as list
df.tail(5)                   # Return last 5 rows (use carefully)

from pyspark.sql.functions import col, lit

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

# Add new column
df.withColumn("age_plus_10", col("age") + 10)
df.withColumn("full_name", col("first_name") + " " + col("last_name"))
df.withColumn("status", lit("active"))  # Add constant value

# Rename columns
df.withColumnRenamed("old_name", "new_name")
df.toDF("new_name1", "new_name2")  # Rename all columns

# Drop columns
df.drop("column1")
df.drop("column1", "column2")

# Cast column type
df.withColumn("age", col("age").cast("string"))
df.withColumn("price", col("price").cast("double"))

from pyspark.sql.functions import col

# Filter rows
df.filter(df["age"] > 25)
df.filter(col("age") > 25)
df.where(col("age") > 25)

# Multiple conditions
df.filter((col("age") > 25) & (col("city") == "NYC"))
df.filter((col("age") < 20) | (col("age") > 60))
df.filter(~(col("status") == "inactive"))

# String operations
df.filter(col("name").startswith("A"))
df.filter(col("name").endswith("son"))
df.filter(col("name").contains("li"))

# Null checks
df.filter(col("age").isNull())
df.filter(col("age").isNotNull())

# In operator
df.filter(col("city").isin(["NYC", "LA", "SF"]))

from pyspark.sql.functions import col, desc, asc

# Sort ascending
df.orderBy("age")
df.sort("age")
df.orderBy(col("age").asc())

# Sort descending
df.orderBy(col("age").desc())
df.sort(col("age").desc())

# Multiple columns
df.orderBy(["age", "name"])
df.orderBy(col("age").desc(), col("name").asc())

from pyspark.sql.functions import count, sum, avg, min, max, countDistinct

# Group by
df.groupBy("department").count()
df.groupBy("department").agg(avg("salary"), max("age"))

# Multiple aggregations
df.groupBy("department").agg(
    count("*").alias("total_employees"),
    avg("salary").alias("avg_salary"),
    max("salary").alias("max_salary"),
    min("salary").alias("min_salary")
)

# Without groupBy (global aggregation)
df.agg(avg("salary"), max("age"))
df.select(avg("salary"), count("*"))

# Distinct count
df.agg(countDistinct("department"))

    Join Types Flow:

    ┌──────────┐        ┌──────────┐
    │   df1    │        │   df2    │
    └────┬─────┘        └─────┬────┘
         │                    │
         └──────────┬─────────┘
                    │
           ┌────────┴────────┐
           │  Join Operation │
           └────────┬────────┘
                    │
        ┌───────────┼───────────┐
        │           │           │
        ↓           ↓           ↓
    ┌───────┐  ┌────────┐  ┌──────┐
    │ inner │  │  left  │  │right │
    └───────┘  │ outer  │  │outer │
               └────────┘  └──────┘
                    │
                    ↓
              ┌──────────┐
              │   full   │
              │  outer   │
              └──────────┘

# Inner join (default)
df1.join(df2, df1.id == df2.id)
df1.join(df2, "id")  # If column name is same

# Left outer join
df1.join(df2, df1.id == df2.id, "left")
df1.join(df2, df1.id == df2.id, "left_outer")

# Right outer join
df1.join(df2, df1.id == df2.id, "right")
df1.join(df2, df1.id == df2.id, "right_outer")

# Full outer join
df1.join(df2, df1.id == df2.id, "full")
df1.join(df2, df1.id == df2.id, "full_outer")

# Left semi join (like SQL IN)
df1.join(df2, df1.id == df2.id, "left_semi")

# Left anti join (like SQL NOT IN)
df1.join(df2, df1.id == df2.id, "left_anti")

# Cross join (Cartesian product)
df1.crossJoin(df2)

# Multiple join conditions
df1.join(df2, (df1.id == df2.id) & (df1.date == df2.date))

# Union (combines all rows, includes duplicates)
df1.union(df2)

# Union by name (matches by column name)
df1.unionByName(df2)

# Intersection (rows in both)
df1.intersect(df2)

# Subtract (rows in df1 but not in df2)
df1.subtract(df2)

# Distinct (remove duplicates)
df.distinct()
df.dropDuplicates()
df.dropDuplicates(["name", "age"])  # Based on specific columns

from pyspark.sql import StorageLevel

# Sampling
df.sample(withReplacement=False, fraction=0.5, seed=42)
df.sample(0.3)  # 30% sample

# Random split
train_df, test_df = df.randomSplit([0.8, 0.2], seed=42)

# Limit
df.limit(10)

# Pivoting
df.groupBy("date").pivot("category").sum("amount")
df.groupBy("date").pivot("category", ["A", "B"]).sum("amount")

# Rollup (hierarchical aggregation)
df.rollup("country", "city").agg(sum("sales"))

# Cube (multi-dimensional aggregation)
df.cube("year", "quarter").agg(sum("revenue"))

# Caching
df.cache()                              # Cache in memory
df.persist(StorageLevel.MEMORY_AND_DISK)  # Custom storage
df.unpersist()                          # Remove from cache
df.persist().count()                    # Cache and trigger computation

# Drop rows with nulls
df.na.drop()                      # Drop if any null
df.na.drop(how="all")             # Drop if all nulls
df.na.drop(subset=["age", "name"]) # Drop if null in specific columns

# Fill nulls
df.na.fill(0)                     # Fill all nulls with 0
df.na.fill({"age": 0, "name": "Unknown"})  # Fill by column
df.fillna(0)                      # Alternative syntax

# Replace values
df.na.replace(["old1", "old2"], ["new1", "new2"])
df.replace(to_replace={"old": "new"}, subset=["column"])

from pyspark.sql.functions import (
    col, lit, when, coalesce, concat, concat_ws,
    lower, upper, trim, length, substring, regexp_replace,
    split, explode, array_contains,
    year, month, dayofmonth, current_date, current_timestamp,
    datediff, date_add, date_sub, to_date, to_timestamp,
    round, ceil, floor, abs, sqrt, pow,
    md5, sha1, sha2, base64, unbase64
)

# String functions
df.withColumn("lower_name", lower(col("name")))
df.withColumn("upper_name", upper(col("name")))
df.withColumn("trimmed", trim(col("name")))
df.withColumn("name_length", length(col("name")))
df.withColumn("first_3", substring(col("name"), 1, 3))
df.withColumn("cleaned", regexp_replace(col("text"), "[^a-zA-Z]", ""))
df.withColumn("full_name", concat(col("first"), lit(" "), col("last")))
df.withColumn("full_name", concat_ws(" ", col("first"), col("last")))

# Array functions
df.withColumn("words", split(col("sentence"), " "))
df.withColumn("word", explode(col("words")))
df.withColumn("has_item", array_contains(col("items"), "target"))

# Date functions
df.withColumn("year", year(col("date")))
df.withColumn("month", month(col("date")))
df.withColumn("day", dayofmonth(col("date")))
df.withColumn("today", current_date())
df.withColumn("now", current_timestamp())
df.withColumn("days_diff", datediff(col("end_date"), col("start_date")))
df.withColumn("next_week", date_add(col("date"), 7))
df.withColumn("last_week", date_sub(col("date"), 7))

# Math functions
df.withColumn("rounded", round(col("value"), 2))
df.withColumn("ceiling", ceil(col("value")))
df.withColumn("floor", floor(col("value")))
df.withColumn("absolute", abs(col("value")))
df.withColumn("square_root", sqrt(col("value")))

# Conditional operations
df.withColumn("category",
    when(col("age") < 18, "minor")
    .when(col("age") < 65, "adult")
    .otherwise("senior")
)

# Null handling
df.withColumn("filled", coalesce(col("col1"), col("col2"), lit("default")))

    UDF Processing Flow:

    ┌───────────────┐
    │ Python UDF    │
    │  (Slow)       │
    └───────┬───────┘
            │
            ↓ Serialization
    ┌───────────────┐
    │  JVM Executor │
    └───────┬───────┘
            │
            ↓ Row by Row
    ┌───────────────┐
    │  Result       │
    └───────────────┘

    Pandas UDF (Faster):

    ┌───────────────┐
    │ Pandas UDF    │
    │ (Vectorized)  │
    └───────┬───────┘
            │
            ↓ Apache Arrow
    ┌───────────────┐
    │  JVM Executor │
    └───────┬───────┘
            │
            ↓ Batch Processing
    ┌───────────────┐
    │  Result       │
    └───────────────┘

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, IntegerType, DoubleType

# Simple UDF
def to_upper(s):
    if s is not None:
        return s.upper()
    return None

to_upper_udf = udf(to_upper, StringType())
df = df.withColumn("upper_name", to_upper_udf(col("name")))

# UDF with decorator
@udf(returnType=IntegerType())
def calculate_age(birth_year):
    from datetime import datetime
    current_year = datetime.now().year
    return current_year - birth_year

df = df.withColumn("age", calculate_age(col("birth_year")))

# UDF with multiple arguments
@udf(returnType=DoubleType())
def calculate_discount(price, discount_pct):
    if price and discount_pct:
        return price * (1 - discount_pct / 100)
    return price

df = df.withColumn("final_price", calculate_discount(col("price"), col("discount")))

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import StringType, DoubleType
import pandas as pd

# Series to Series
@pandas_udf(StringType())
def to_upper_pandas(series: pd.Series) -> pd.Series:
    return series.str.upper()

df = df.withColumn("upper_name", to_upper_pandas(col("name")))

# Multiple columns
@pandas_udf(DoubleType())
def calculate_discount(price: pd.Series, discount: pd.Series) -> pd.Series:
    return price * (1 - discount / 100)

df = df.withColumn("final_price", calculate_discount(col("price"), col("discount")))

# Iterator of Series (for large data)
from typing import Iterator

@pandas_udf(DoubleType())
def complex_calculation(iterator: Iterator[pd.Series]) -> Iterator[pd.Series]:
    for batch in iterator:
        # Process batch
        yield batch * 2

df = df.withColumn("doubled", complex_calculation(col("value")))

# Register UDF for SQL use
spark.udf.register("to_upper_sql", to_upper, StringType())
df.createOrReplaceTempView("my_table")
result = spark.sql("SELECT to_upper_sql(name) FROM my_table")

# Use built-in functions when possible (much faster)
# Instead of UDF for upper:
df.withColumn("upper_name", upper(col("name")))  # Preferred

# Cache if using UDF multiple times
df_cached = df.cache()
df_cached = df_cached.withColumn("col1", my_udf(col("x")))
df_cached = df_cached.withColumn("col2", my_udf(col("y")))

from pyspark.sql.functions import (
    count, sum, avg, max, min, mean, stddev,
    collect_list, collect_set, countDistinct
)

# Single column grouping
df.groupBy("department").count()
df.groupBy("department").sum("salary")
df.groupBy("department").avg("salary")
df.groupBy("department").max("salary")
df.groupBy("department").min("salary")

# Multiple column grouping
df.groupBy("department", "city").count()

# Multiple aggregations
df.groupBy("department").agg(
    count("*").alias("total_employees"),
    avg("salary").alias("avg_salary"),
    max("salary").alias("max_salary"),
    min("salary").alias("min_salary"),
    sum("bonus").alias("total_bonus"),
    countDistinct("employee_id").alias("unique_employees")
)

# Collect values into list/set
df.groupBy("department").agg(
    collect_list("employee_name").alias("employees"),
    collect_set("skill").alias("unique_skills")
)

# Group and filter
df.groupBy("department") \
    .agg(avg("salary").alias("avg_salary")) \
    .filter(col("avg_salary") > 50000)

    Window Function Partitioning:

    ┌────────────────────────────────────┐
    │       Full DataFrame               │
    └────────────┬───────────────────────┘
                 │
         Partition By
                 │
    ┌────────────┼────────────┐
    │            │            │
    ↓            ↓            ↓
┌─────────┐  ┌─────────┐  ┌─────────┐
│ Dept A  │  │ Dept B  │  │ Dept C  │
│ Window  │  │ Window  │  │ Window  │
└────┬────┘  └────┬────┘  └────┬────┘
     │            │            │
  Order By     Order By     Order By
     │            │            │
     ↓            ↓            ↓
  Ranking     Ranking     Ranking

from pyspark.sql import Window
from pyspark.sql.functions import (
    row_number, rank, dense_rank, percent_rank,
    lag, lead, first, last,
    sum, avg, max, min, count
)

# Define window specifications
# Partition by department, order by salary descending
windowSpec = Window.partitionBy("department").orderBy(col("salary").desc())

# Ranking functions
df.withColumn("row_num", row_number().over(windowSpec))
df.withColumn("rank", rank().over(windowSpec))
df.withColumn("dense_rank", dense_rank().over(windowSpec))
df.withColumn("percent_rank", percent_rank().over(windowSpec))

# Analytic functions
df.withColumn("prev_salary", lag("salary", 1).over(windowSpec))
df.withColumn("next_salary", lead("salary", 1).over(windowSpec))
df.withColumn("first_salary", first("salary").over(windowSpec))
df.withColumn("last_salary", last("salary").over(windowSpec))

# Aggregate window functions
df.withColumn("dept_total_salary", sum("salary").over(windowSpec))
df.withColumn("dept_avg_salary", avg("salary").over(windowSpec))
df.withColumn("dept_max_salary", max("salary").over(windowSpec))

# Running totals
runningTotalWindow = Window \
    .partitionBy("department") \
    .orderBy("date") \
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)

df.withColumn("running_total", sum("amount").over(runningTotalWindow))

# Moving average (last 3 rows)
movingAvgWindow = Window \
    .partitionBy("department") \
    .orderBy("date") \
    .rowsBetween(-2, 0)  # 2 rows before and current row

df.withColumn("moving_avg", avg("value").over(movingAvgWindow))

# Range-based window (value-based, not row-based)
rangeWindow = Window \
    .partitionBy("department") \
    .orderBy("salary") \
    .rangeBetween(-1000, 1000)  # Within 1000 of current salary

df.withColumn("peers_count", count("*").over(rangeWindow))

# No partitioning (entire dataset)
globalWindow = Window.orderBy("date")
df.withColumn("global_rank", rank().over(globalWindow))

# Top N per group using window functions
windowSpec = Window.partitionBy("department").orderBy(col("salary").desc())
top_earners = df.withColumn("rank", rank().over(windowSpec)) \
    .filter(col("rank") <= 3)

df.createOrReplaceTempView("my_table")

result_df = spark.sql("SELECT Name, Age FROM my_table WHERE Age > 25")
result_df.show()

    ┌─────────────────────────────────────────┐
    │                RDD                      │
    │  • Low-level API                        │
    │  • No schema                            │
    │  • Type-safe (in Scala)                 │
    │  • No optimization                      │
    │  • Manual optimization needed           │
    └─────────────────────────────────────────┘
                    VS
    ┌─────────────────────────────────────────┐
    │             DataFrame                   │
    │  • High-level API                       │
    │  • Schema-based                         │
    │  • Catalyst optimizer                   │
    │  • Tungsten execution                   │
    │  • Better performance                   │
    └─────────────────────────────────────────┘

# From Python collection
data = [1, 2, 3, 4, 5]
rdd = sc.parallelize(data)

# With specific number of partitions
rdd = sc.parallelize(data, numSlices=4)

# From text file
rdd = sc.textFile("path/to/file.txt")
rdd = sc.textFile("path/to/directory/*.txt")

# From whole text files (returns filename, content pairs)
rdd = sc.wholeTextFiles("path/to/directory")

# From sequence file
rdd = sc.sequenceFile("path/to/file")

# Empty RDD
empty_rdd = sc.emptyRDD()

# From range
rdd = sc.parallelize(range(1, 100))

    Transformation Pipeline:

    ┌──────────┐
    │   RDD    │
    └────┬─────┘
         │
         ↓ map()
    ┌────────────┐
    │ Transformed│
    └────┬───────┘
         │
         ↓ filter()
    ┌────────────┐
    │  Filtered  │
    └────┬───────┘
         │
         ↓ reduce() (Action)
    ┌────────────┐
    │  Execute   │
    │  Pipeline  │
    └────────────┘

# Map: Apply function to each element
rdd.map(lambda x: x * 2)
rdd.map(lambda x: (x, 1))  # Create pairs

# FlatMap: Map and flatten results
rdd.flatMap(lambda x: x.split())
rdd.flatMap(lambda x: range(x))

# Filter: Keep elements matching condition
rdd.filter(lambda x: x > 2)
rdd.filter(lambda x: x % 2 == 0)

# MapPartitions: Apply function to each partition
def process_partition(iterator):
    yield sum(iterator)
rdd.mapPartitions(process_partition)

# MapPartitionsWithIndex: Include partition index
def process_with_index(index, iterator):
    yield (index, sum(iterator))
rdd.mapPartitionsWithIndex(process_with_index)

# Distinct: Remove duplicates
rdd.distinct()

# Sample: Random sample
rdd.sample(withReplacement=False, fraction=0.5, seed=42)

# Union: Combine RDDs
rdd1.union(rdd2)

# Intersection: Common elements
rdd1.intersection(rdd2)

# Subtract: Elements in rdd1 but not rdd2
rdd1.subtract(rdd2)

# Cartesian: Cartesian product
rdd1.cartesian(rdd2)

# Sort by value
rdd.sortBy(lambda x: x, ascending=True)

# Repartition: Change number of partitions (shuffle)
rdd.repartition(10)

# Coalesce: Reduce partitions (no shuffle)
rdd.coalesce(1)

# Partition by custom function
rdd.partitionBy(numPartitions=10, partitionFunc=lambda x: x % 10)

# Zip: Combine element-wise
rdd1.zip(rdd2)  # Same partitioning required

# ZipWithIndex: Add indices
rdd.zipWithIndex()

# ZipWithUniqueId: Add unique IDs
rdd.zipWithUniqueId()

# Collect: Return all elements (use carefully!)
rdd.collect()

# Count: Number of elements
rdd.count()

# First: First element
rdd.first()

# Take: First N elements
rdd.take(5)

# Top: Top N elements
rdd.top(5)

# TakeOrdered: N smallest elements
rdd.takeOrdered(5)
rdd.takeOrdered(5, key=lambda x: -x)  # Largest

# TakeSample: Random sample
rdd.takeSample(withReplacement=False, num=5, seed=42)

# Reduce: Aggregate elements
rdd.reduce(lambda x, y: x + y)
rdd.reduce(lambda x, y: x if x > y else y)  # Max

# Fold: Like reduce with initial value
rdd.fold(0, lambda x, y: x + y)

# Aggregate: Different types for accumulator and elements
rdd.aggregate(
    (0, 0),  # Initial value
    lambda acc, value: (acc[0] + value, acc[1] + 1),  # Seq op
    lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])  # Comb op
)

# Foreach: Apply function to each element (no return)
rdd.foreach(lambda x: print(x))

# ForeachPartition: Apply function to each partition
def print_partition(iterator):
    for item in iterator:
        print(item)
rdd.foreachPartition(print_partition)

# CountByValue: Count occurrences of each value
rdd.countByValue()

# SaveAsTextFile: Save to files
rdd.saveAsTextFile("path/to/output")

# SaveAsSequenceFile: Save as Hadoop SequenceFile
rdd.saveAsSequenceFile("path/to/output")

# Create pair RDD
words = sc.parallelize(["hello", "world", "hello"])
pairs = words.map(lambda x: (x, 1))

# ReduceByKey: Combine values for each key
word_counts = pairs.reduceByKey(lambda x, y: x + y)

# GroupByKey: Group all values for each key (avoid if possible)
pairs.groupByKey().mapValues(list)

# AggregateByKey: Aggregate with initial value
pairs.aggregateByKey(
    0,  # Initial value
    lambda acc, value: acc + value,  # Seq func
    lambda acc1, acc2: acc1 + acc2   # Comb func
)

# FoldByKey: Fold with initial value
pairs.foldByKey(0, lambda x, y: x + y)

# CombineByKey: Most general aggregation
def create_combiner(x):
    return (x, 1)

def merge_value(acc, x):
    return (acc[0] + x, acc[1] + 1)

def merge_combiners(acc1, acc2):
    return (acc1[0] + acc2[0], acc1[1] + acc2[1])

pairs.combineByKey(create_combiner, merge_value, merge_combiners)

# MapValues: Transform only values
pairs.mapValues(lambda x: x * 2)

# FlatMapValues: FlatMap only values
pairs.flatMapValues(lambda x: range(x))

# Keys: Get all keys
pairs.keys()

# Values: Get all values
pairs.values()

# SortByKey: Sort by key
pairs.sortByKey(ascending=True)

# Join: Inner join
rdd1 = sc.parallelize([("a", 1), ("b", 2)])
rdd2 = sc.parallelize([("a", 3), ("c", 4)])
rdd1.join(rdd2)  # Result: [("a", (1, 3))]

# LeftOuterJoin
rdd1.leftOuterJoin(rdd2)  # Result: [("a", (1, Some(3))), ("b", (2, None))]

# RightOuterJoin
rdd1.rightOuterJoin(rdd2)  # Result: [("a", (Some(1), 3)), ("c", (None, 4))]

# FullOuterJoin
rdd1.fullOuterJoin(rdd2)

# Cogroup: Group from multiple RDDs
rdd1.cogroup(rdd2)

# Pair RDD actions
pairs.countByKey()        # Count values per key
pairs.collectAsMap()      # Collect as dictionary
pairs.lookup("hello")     # Get all values for key

from pyspark import StorageLevel

# Cache in memory
rdd.cache()  # Same as persist(StorageLevel.MEMORY_ONLY)

# Persist with storage level
rdd.persist(StorageLevel.MEMORY_AND_DISK)
rdd.persist(StorageLevel.MEMORY_ONLY)
rdd.persist(StorageLevel.DISK_ONLY)
rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)  # Serialized

# Check if persisted
rdd.is_cached

# Unpersist
rdd.unpersist()

# Get number of partitions
rdd.getNumPartitions()

# Checkpoint (for lineage truncation)
sc.setCheckpointDir("path/to/checkpoint")
rdd.checkpoint()

    Write Mode Decision Tree:

    ┌──────────────────┐
    │  Target Exists?  │
    └────────┬─────────┘
             │
        ┌────┴────┐
        ↓         ↓
      Yes        No
        │         │
        ↓         ↓
    ┌───────┐  ┌────────┐
    │ Mode? │  │ Create │
    └───┬───┘  └────────┘
        │
    ┌───┴────────────────┐
    │                    │
    ↓                    ↓
┌──────────┐      ┌────────────┐
│overwrite │      │   append   │
└──────────┘      └────────────┘
    │                    │
    ↓                    ↓
┌──────────┐      ┌────────────┐
│ Replace  │      │  Add New   │
│   All    │      │   Rows     │
└──────────┘      └────────────┘

    ↓ error         ↓ ignore
┌──────────┐   ┌────────────┐
│  Throw   │   │  Skip if   │
│  Error   │   │  Exists    │
└──────────┘   └────────────┘

# Basic write
df.write.csv("path/to/output", header=True, mode="overwrite")

# With options
df.write \
    .format("csv") \
    .option("header", "true") \
    .option("sep", ",") \
    .option("quote", '"') \
    .option("escape", "\\") \
    .option("compression", "gzip") \
    .option("dateFormat", "yyyy-MM-dd") \
    .mode("overwrite") \
    .save("path/to/output")

# Single file output
df.coalesce(1).write.csv("path/to/output", header=True)

# Basic write
df.write.json("path/to/output", mode="overwrite")

# With options
df.write \
    .format("json") \
    .option("compression", "gzip") \
    .option("dateFormat", "yyyy-MM-dd") \
    .mode("overwrite") \
    .save("path/to/output")

# Basic write
df.write.parquet("path/to/output", mode="overwrite")

# With compression
df.write \
    .format("parquet") \
    .option("compression", "snappy")  # gzip, snappy, lzo, none
    .mode("overwrite") \
    .save("path/to/output")

# Partitioned write
df.write \
    .partitionBy("year", "month") \
    .parquet("path/to/output", mode="overwrite")

# With bucketing
df.write \
    .bucketBy(10, "user_id") \
    .sortBy("timestamp") \
    .saveAsTable("user_events")

# Basic write
df.write.orc("path/to/output", mode="overwrite")

# With compression
df.write \
    .format("orc") \
    .option("compression", "snappy") \
    .mode("overwrite") \
    .save("path/to/output")

# Basic write
df.write.format("avro").save("path/to/output", mode="overwrite")

# With compression
df.write \
    .format("avro") \
    .option("compression", "snappy") \
    .save("path/to/output", mode="overwrite")

# Basic write
df.write.format("jdbc") \
    .option("url", "jdbc:postgresql://localhost:5432/mydatabase") \
    .option("dbtable", "mytable") \
    .option("user", "myuser") \
    .option("password", "mypassword") \
    .option("driver", "org.postgresql.Driver") \
    .mode("overwrite") \
    .save()

# With batching for better performance
df.write.format("jdbc") \
    .option("url", "jdbc:postgresql://localhost:5432/mydatabase") \
    .option("dbtable", "mytable") \
    .option("user", "myuser") \
    .option("password", "mypassword") \
    .option("batchsize", "10000") \
    .option("isolationLevel", "NONE") \
    .mode("append") \
    .save()

# Truncate and insert
df.write.format("jdbc") \
    .option("url", "jdbc:postgresql://localhost:5432/mydatabase") \
    .option("dbtable", "mytable") \
    .option("user", "myuser") \
    .option("password", "mypassword") \
    .option("truncate", "true") \
    .mode("overwrite") \
    .save()

# Basic write
df.write.format("delta").mode("overwrite").save("path/to/delta_table")

# Append mode
df.write.format("delta").mode("append").save("path/to/delta_table")

# With partitioning
df.write \
    .format("delta") \
    .partitionBy("date") \
    .mode("overwrite") \
    .save("path/to/delta_table")

# Write as managed table
df.write.format("delta").saveAsTable("my_delta_table")

# Merge (upsert) operation
from delta.tables import DeltaTable

deltaTable = DeltaTable.forPath(spark, "path/to/delta_table")
deltaTable.alias("target").merge(
    df.alias("source"),
    "target.id = source.id"
).whenMatchedUpdate(set={
    "value": "source.value",
    "updated_at": "source.updated_at"
}).whenNotMatchedInsert(values={
    "id": "source.id",
    "value": "source.value",
    "updated_at": "source.updated_at"
}).execute()

# Save as managed table
df.write.saveAsTable("database.tablename", mode="overwrite")

# Save as external table
df.write \
    .option("path", "hdfs://path/to/data") \
    .saveAsTable("database.tablename", mode="overwrite")

# Partitioned table
df.write \
    .partitionBy("year", "month") \
    .saveAsTable("database.tablename", mode="overwrite")

# Insert into existing table
df.write.insertInto("database.tablename", overwrite=True)

# Overwrite: Replace existing data
df.write.mode("overwrite").parquet("path/to/output")

# Append: Add to existing data
df.write.mode("append").parquet("path/to/output")

# Error (default): Throw error if exists
df.write.mode("error").parquet("path/to/output")
df.write.mode("errorifexists").parquet("path/to/output")

# Ignore: Skip if exists
df.write.mode("ignore").parquet("path/to/output")

# Save as text file
rdd.saveAsTextFile("path/to/output")

# With compression
rdd.saveAsTextFile("path/to/output", compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec")

# Save as pickle file
rdd.saveAsPickleFile("path/to/output")

# Save as sequence file (Hadoop format)
rdd.saveAsSequenceFile("path/to/output")

# Save as Hadoop file with custom format
rdd.saveAsHadoopFile(
    "path/to/output",
    "org.apache.hadoop.mapred.TextOutputFormat"
)

# Convert to DataFrame and write
rdd.toDF(["column1", "column2"]).write.parquet("path/to/output")

# Partition by column values
df.write \
    .partitionBy("year", "month", "day") \
    .parquet("path/to/output")

# Control number of output files
df.repartition(10).write.parquet("path/to/output")

# Single output file (use carefully!)
df.coalesce(1).write.parquet("path/to/output")

# Partition and control file size
df.repartition("year", "month") \
    .write \
    .partitionBy("year", "month") \
    .parquet("path/to/output")

df.write.saveAsTable("my_table")

spark.sql("CREATE TABLE my_table (name STRING, age INT) USING parquet")

df.write.insertInto("my_table")

spark.sql("INSERT INTO my_table VALUES ('Alice', 30)")

spark.sql("SELECT * FROM my_table").show()

spark.sql("SELECT * FROM my_table WHERE age > 25").show()

spark.sql("SELECT name, AVG(age) FROM my_table GROUP BY name").show()

spark.sql("SELECT * FROM table1 JOIN table2 ON table1.key = table2.key").show()

spark.sql("""
SELECT
    name,
    age,
    RANK() OVER (ORDER BY age DESC) as age_rank
FROM
    my_table
""").show()

    ┌───────────────┐
    │   Raw Data    │
    └───────┬───────┘
            │
            ↓
    ┌───────────────┐
    │ StringIndexer │  (Categorical → Numeric)
    └───────┬───────┘
            │
            ↓
    ┌───────────────┐
    │VectorAssembler│  (Combine Features)
    └───────┬───────┘
            │
            ↓
    ┌───────────────┐
    │ StandardScaler│  (Normalize)
    └───────┬───────┘
            │
            ↓
    ┌───────────────┐
    │  Train/Test   │
    │     Split     │
    └───────┬───────┘
            │
        ┌───┴───┐
        ↓       ↓
    ┌──────┐ ┌──────┐
    │Train │ │ Test │
    └──┬───┘ └───┬──┘
       │         │
       ↓         │
    ┌──────┐    │
    │Model │    │
    └──┬───┘    │
       │        │
       └────────┴──→ Predictions

from pyspark.ml.feature import (
    StringIndexer, VectorAssembler, StandardScaler,
    MinMaxScaler, Normalizer, OneHotEncoder,
    Tokenizer, StopWordsRemover, CountVectorizer,
    IDF, PCA, Imputer
)

# StringIndexer: Convert categories to indices
indexer = StringIndexer(inputCol="category", outputCol="categoryIndex")
indexed_df = indexer.fit(df).transform(df)

# Handle unseen labels
indexer = StringIndexer(
    inputCol="category",
    outputCol="categoryIndex",
    handleInvalid="keep"  # "skip" or "error"
)

# OneHotEncoder: Convert indices to binary vectors
encoder = OneHotEncoder(inputCols=["categoryIndex"], outputCols=["categoryVec"])
encoded_df = encoder.fit(indexed_df).transform(indexed_df)

# VectorAssembler: Combine features into vector
assembler = VectorAssembler(
    inputCols=["feature1", "feature2", "feature3"],
    outputCol="features",
    handleInvalid="skip"  # "keep" or "error"
)
output_df = assembler.transform(df)

# Imputer: Fill missing values
imputer = Imputer(
    inputCols=["age", "income"],
    outputCols=["age_imputed", "income_imputed"],
    strategy="mean"  # "median" or "mode"
)
imputer_model = imputer.fit(df)
imputed_df = imputer_model.transform(df)

# StandardScaler: Standardize features (mean=0, std=1)
scaler = StandardScaler(
    inputCol="features",
    outputCol="scaledFeatures",
    withStd=True,
    withMean=True
)
scaler_model = scaler.fit(df)
scaled_df = scaler_model.transform(df)

# MinMaxScaler: Scale to range [0, 1]
min_max_scaler = MinMaxScaler(
    inputCol="features",
    outputCol="scaledFeatures",
    min=0.0,
    max=1.0
)
min_max_model = min_max_scaler.fit(df)
scaled_df = min_max_model.transform(df)

# Normalizer: Normalize to unit norm
normalizer = Normalizer(
    inputCol="features",
    outputCol="normFeatures",
    p=2.0  # L2 norm (p=1.0 for L1)
)
normalized_df = normalizer.transform(df)

# PCA: Dimensionality reduction
pca = PCA(
    k=3,  # Number of principal components
    inputCol="features",
    outputCol="pcaFeatures"
)
pca_model = pca.fit(df)
pca_df = pca_model.transform(df)

from pyspark.ml.feature import (
    Tokenizer, RegexTokenizer, StopWordsRemover,
    CountVectorizer, HashingTF, IDF, Word2Vec, NGram
)

# Tokenizer: Split text into words
tokenizer = Tokenizer(inputCol="text", outputCol="words")
tokenized_df = tokenizer.transform(df)

# RegexTokenizer: Tokenize with regex
regex_tokenizer = RegexTokenizer(
    inputCol="text",
    outputCol="words",
    pattern="\\W"  # Split on non-word characters
)

# StopWordsRemover: Remove common words
remover = StopWordsRemover(inputCol="words", outputCol="filtered")
removed_df = remover.transform(tokenized_df)

# CountVectorizer: Bag of words
cv = CountVectorizer(
    inputCol="filtered",
    outputCol="rawFeatures",
    vocabSize=10000,
    minDF=2.0  # Minimum document frequency
)
cv_model = cv.fit(removed_df)
cv_df = cv_model.transform(removed_df)

# TF-IDF: Term frequency - inverse document frequency
# First HashingTF
from pyspark.ml.feature import HashingTF
hashing_tf = HashingTF(inputCol="filtered", outputCol="rawFeatures", numFeatures=10000)
tf_df = hashing_tf.transform(removed_df)

# Then IDF
idf = IDF(inputCol="rawFeatures", outputCol="features")
idf_model = idf.fit(tf_df)
tfidf_df = idf_model.transform(tf_df)

# Word2Vec: Word embeddings
word2vec = Word2Vec(
    vectorSize=100,
    minCount=5,
    inputCol="filtered",
    outputCol="features"
)
w2v_model = word2vec.fit(removed_df)
w2v_df = w2v_model.transform(removed_df)

# NGram: Generate n-grams
ngram = NGram(n=2, inputCol="words", outputCol="ngrams")
ngram_df = ngram.transform(tokenized_df)

    Classification Model Selection:

    ┌──────────────────┐
    │  Problem Type?   │
    └────────┬─────────┘
             │
    ┌────────┴────────┐
    │                 │
    ↓                 ↓
┌─────────┐      ┌─────────┐
│ Binary  │      │  Multi  │
│  Class  │      │  Class  │
└────┬────┘      └────┬────┘
     │                │
     ↓                ↓
┌──────────────────────────┐
│ Linear Separable?        │
└────────┬─────────────────┘
         │
    ┌────┴─────┐
    ↓          ↓
  Yes         No
    │          │
    ↓          ↓
┌────────┐  ┌──────────┐
│Logistic│  │Random    │
│Regress │  │Forest/   │
│        │  │GBT/SVM   │
└────────┘  └──────────┘

from pyspark.ml.classification import LogisticRegression

# Binary classification
lr = LogisticRegression(
    featuresCol="features",
    labelCol="label",
    maxIter=100,
    regParam=0.01,        # Regularization parameter
    elasticNetParam=0.5,  # 0=L2, 1=L1, 0.5=mix
    threshold=0.5         # Classification threshold
)
model = lr.fit(training_data)
predictions = model.transform(test_data)

# Multi-class classification
lr_multi = LogisticRegression(
    featuresCol="features",
    labelCol="label",
    family="multinomial"
)

# Access model parameters
print(f"Coefficients: {model.coefficients}")
print(f"Intercept: {model.intercept}")

# Summary
training_summary = model.summary
print(f"Accuracy: {training_summary.accuracy}")
print(f"Area under ROC: {training_summary.areaUnderROC}")

from pyspark.ml.classification import DecisionTreeClassifier

dt = DecisionTreeClassifier(
    featuresCol="features",
    labelCol="label",
    maxDepth=5,           # Maximum depth
    maxBins=32,           # Number of bins for continuous features
    impurity="gini",      # "gini" or "entropy"
    minInstancesPerNode=1 # Minimum instances per node
)
model = dt.fit(training_data)
predictions = model.transform(test_data)

# Feature importance
print(f"Feature importances: {model.featureImportances}")

# Tree structure
print(model.toDebugString)

from pyspark.ml.classification import RandomForestClassifier

rf = RandomForestClassifier(
    featuresCol="features",
    labelCol="label",
    numTrees=100,         # Number of trees
    maxDepth=5,
    maxBins=32,
    subsamplingRate=1.0,  # Fraction of training data
    featureSubsetStrategy="auto",  # "all", "sqrt", "log2", "onethird"
    seed=42
)
model = rf.fit(training_data)
predictions = model.transform(test_data)

# Feature importance
print(f"Feature importances: {model.featureImportances}")

# Access individual trees
print(f"Number of trees: {model.numTrees}")
print(f"Total nodes: {model.totalNumNodes}")

from pyspark.ml.classification import GBTClassifier

gbt = GBTClassifier(
    featuresCol="features",
    labelCol="label",
    maxIter=20,           # Number of boosting iterations
    maxDepth=5,
    stepSize=0.1,         # Learning rate
    subsamplingRate=1.0,
    featureSubsetStrategy="all"
)
model = gbt.fit(training_data)
predictions = model.transform(test_data)

# Feature importance
print(f"Feature importances: {model.featureImportances}")

from pyspark.ml.classification import NaiveBayes

nb = NaiveBayes(
    featuresCol="features",
    labelCol="label",
    smoothing=1.0,        # Laplace smoothing
    modelType="multinomial"  # "multinomial", "bernoulli", "gaussian"
)
model = nb.fit(training_data)
predictions = model.transform(test_data)

from pyspark.ml.classification import LinearSVC

svm = LinearSVC(
    featuresCol="features",
    labelCol="label",
    maxIter=100,
    regParam=0.01,
    threshold=0.0
)
model = svm.fit(training_data)
predictions = model.transform(test_data)

from pyspark.ml.classification import MultilayerPerceptronClassifier

# Define network architecture
# [input_size, hidden_layer1, hidden_layer2, ..., output_size]
layers = [10, 20, 15, 3]  # 10 features, 2 hidden layers, 3 classes

mlp = MultilayerPerceptronClassifier(
    layers=layers,
    featuresCol="features",
    labelCol="label",
    maxIter=100,
    blockSize=128,        # Block size for stacking input data
    seed=42
)
model = mlp.fit(training_data)
predictions = model.transform(test_data)

from pyspark.ml.regression import LinearRegression

lr = LinearRegression(
    featuresCol="features",
    labelCol="label",
    maxIter=100,
    regParam=0.01,        # L2 regularization
    elasticNetParam=0.0,  # 0=L2 (Ridge), 1=L1 (Lasso)
    standardization=True,
    fitIntercept=True
)
model = lr.fit(training_data)
predictions = model.transform(test_data)

# Model statistics
print(f"Coefficients: {model.coefficients}")
print(f"Intercept: {model.intercept}")
print(f"RMSE: {model.summary.rootMeanSquaredError}")
print(f"R2: {model.summary.r2}")
print(f"Mean Absolute Error: {model.summary.meanAbsoluteError}")

from pyspark.ml.regression import DecisionTreeRegressor

dt = DecisionTreeRegressor(
    featuresCol="features",
    labelCol="label",
    maxDepth=5,
    maxBins=32,
    minInstancesPerNode=1,
    varianceCol="variance"  # Output variance column
)
model = dt.fit(training_data)
predictions = model.transform(test_data)

# Feature importance
print(f"Feature importances: {model.featureImportances}")

from pyspark.ml.regression import RandomForestRegressor

rf = RandomForestRegressor(
    featuresCol="features",
    labelCol="label",
    numTrees=100,
    maxDepth=5,
    maxBins=32,
    subsamplingRate=1.0,
    featureSubsetStrategy="auto",
    seed=42
)
model = rf.fit(training_data)
predictions = model.transform(test_data)

# Feature importance
print(f"Feature importances: {model.featureImportances}")

from pyspark.ml.regression import GBTRegressor

gbt = GBTRegressor(
    featuresCol="features",
    labelCol="label",
    maxIter=20,
    maxDepth=5,
    stepSize=0.1,
    lossType="squared",   # "squared", "absolute"
    subsamplingRate=1.0
)
model = gbt.fit(training_data)
predictions = model.transform(test_data)

# Feature importance
print(f"Feature importances: {model.featureImportances}")

from pyspark.ml.regression import GeneralizedLinearRegression

glr = GeneralizedLinearRegression(
    featuresCol="features",
    labelCol="label",
    family="gaussian",     # "gaussian", "binomial", "poisson", "gamma"
    link="identity",       # Link function
    maxIter=100,
    regParam=0.01
)
model = glr.fit(training_data)
predictions = model.transform(test_data)

from pyspark.ml.regression import IsotonicRegression

iso = IsotonicRegression(
    featuresCol="features",
    labelCol="label",
    isotonic=True  # True for ascending, False for descending
)
model = iso.fit(training_data)
predictions = model.transform(test_data)