How do you optimize a PySpark job?

Optimizing a PySpark job involves several layers of tuning. First, ensure data is partitioned appropriately: use coalesce or repartition to match the number of executors and avoid small files. Second, cache or persist intermediate RDDs/DataFrames that are reused, but only when they fit in memory. Third, broadcast small lookup tables instead of performing shuffle joins; use broadcast joins when one side is < 100MB. Fourth, avoid wide transformations that trigger shuffles, such as groupByKey; prefer reduceByKey or aggregateByKey. Fifth, enable Tungsten and whole-stage code generation by setting spark.sql.codegen.wholeStage=true. Sixth, tune the shuffle partitions (spark.sql.shuffle.partitions) to a value close to the number of cores. Seventh, use efficient file formats like Parquet or ORC with compression and column pruning. Finally, monitor the job with the Spark UI, look for stages with long tasks or high GC, and adjust executor memory, cores, and spark.executor.memoryOverhead accordingly. By combining these practices, you can reduce execution time, lower memory usage, and improve overall job stability.