Fill Nulls:

df.na.fill(0).show() (fill all numeric nulls with 0)
df.na.fill("N/A", subset=["string_col"]).show()
df.na.fill({"col1": 0, "col2": "unknown"}).show()

Using mean/median:

mean_age = df.agg(F.mean("age")).collect()[0][0]
df.na.fill(mean_age, subset=["age"]).show()

9. User-Defined Functions (UDFs)

Import: from pyspark.sql.functions import udf

Define Python function:

def square_udf(x):
    return x * x
square_udf_spark = udf(square_udf, IntegerType()) # Or StringType(), DoubleType(), etc.

Apply UDF: df.withColumn("squared_value", square_udf_spark(df.value_col)).show()

Register UDF (for SQL):

spark.udf.register("square_sql", square_udf, IntegerType())
spark.sql("SELECT value_col, square_sql(value_col) FROM my_table").show()

Pandas UDF (Vectorized UDFs):

from pyspark.sql.functions import pandas_udf, PandasUDFType
from pyspark.sql.types import LongType

@pandas_udf(LongType(), PandasUDFType.SCALAR)
def pandas_plus_one(series):
    return series + 1
df.withColumn("plus_one", pandas_plus_one(df.value_col)).show()

10. RDD Operations (Resilient Distributed Datasets)

Create RDD: rdd = sc.parallelize([1, 2, 3, 4, 5])
Transformations: (lazy evaluation)
- map(lambda x: x*2)
- filter(lambda x: x > 2)
- flatMap(lambda x: range(1, x+1))
- reduceByKey(lambda a, b: a + b) (for key-value RDDs)
- groupByKey()
- sortByKey()
- union(other_rdd)
- join(other_rdd)
Actions: (trigger computation)
- collect() (get all data to driver - use with caution)
- take(n)
- first()
- count()
- reduce(lambda a, b: a + b)
- foreach(lambda x: print(x))
- saveAsTextFile("path/to/output")
Convert to DataFrame: rdd.toDF(["col1", "col2"])
Convert from DataFrame: df.rdd

11. DataFrames - Write Operations

Write CSV: df.write.csv("path/to/output.csv", header=True, mode="overwrite")
Write JSON: df.write.json("path/to/output.json", mode="append")
Write Parquet: df.write.parquet("path/to/output.parquet", mode="overwrite")
Write ORC: df.write.orc("path/to/output.orc", mode="overwrite")

Write JDBC:

df.write.format("jdbc").options(
    url="jdbc:postgresql://host:port/database",
    dbtable="table_name",
    user="username",
    password="password"
).mode("overwrite").save()

Partition By: df.write.partitionBy("col_name").parquet("path/to/output")
Save As Table: df.write.saveAsTable("my_catalog_table")

12. Machine Learning (MLlib - DataFrames API)

Import: from pyspark.ml.feature import ..., from pyspark.ml.classification import ..., etc.
VectorAssembler: Combines feature columns into a si