Convert complex SQL (CTEs, window functions, subqueries) to production-grade PySpark. Discuss when to use spark.sql() vs. DataFrame API, and the implications for testability, partitioning, and execution predictability.

Two approaches: spark.sql() for direct translation and DataFrame API for programmatic logic. SQL approach: createOrReplaceTempView, run ANSI-like SQL—fast parity, but string-based, harder to unit test, and execution plan less explicit. DataFrame API: composable, testable (pass mock DataFrames), explicit transformations. Example—top 3 products by revenue per region:

from pyspark.sql import functions as F
from pyspark.sql.window import Window
# SQL approach
df.createOrReplaceTempView("sales")
result = spark.sql('''
  WITH revenue AS (
    SELECT region, product_id, SUM(quantity*price) AS revenue
    FROM sales GROUP BY region, product_id
  )
  SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY region ORDER BY revenue DESC) AS rn
    FROM revenue
  ) WHERE rn <= 3
''')
# DataFrame API
revenue = sales.groupBy("region", "product_id").agg(F.sum(F.col("quantity")*F.col("price")).alias("revenue"))
w = Window.partitionBy("region").orderBy(F.col("revenue").desc())
result = revenue.withColumn("rn", F.row_number().over(w)).filter(F.col("rn") <= 3)

Scalability: DataFrame API allows explicit repartition, broadcast hints; SQL relies on optimizer. Cost: SQL is faster to write; DataFrame reduces debugging time and enables CI for logic.