PySpark and SQL

spark.sql("SELECT count(*) AS flights FROM flights").show()

spark.sql("DESCRIBE flights").show(20, truncate=False)
# DESCRIBE reads the metadata and count reads the data, so one is instant and the other is not.

by_carrier = spark.sql("""
    SELECT carrier,
           count(*) AS flights,
           round(avg(dep_delay), 2) AS avg_delay
    FROM flights
    WHERE dep_delay IS NOT NULL
    GROUP BY carrier
    ORDER BY avg_delay DESC
""")

by_carrier.show()

unfiltered = spark.sql("""
    SELECT carrier,
           count(*) AS flights,
           round(avg(dep_delay), 2) AS avg_delay
    FROM flights
    GROUP BY carrier
    ORDER BY avg_delay DESC
""")

unfiltered.show()
# The average ignores nulls whether you filter them or not, so avg_delay does not move.
# The count does : without the filter the two columns describe different sets of flights.

spark.sql("""
    SELECT carrier, count(*) AS flights, round(avg(dep_delay), 2) AS avg_delay
    FROM flights
    GROUP BY carrier
    HAVING count(*) > 100000
    ORDER BY avg_delay DESC
""").show()
# HAVING filters groups where WHERE filters rows.
# In the DataFrame API a filter after groupBy().agg() does the same.

spark.sql("""
    SELECT flight_date, carrier, origin, dest, dep_delay
    FROM flights WHERE dep_delay > 60
""").createOrReplaceTempView("very_late")

print("exists :", spark.catalog.tableExists("very_late"))
spark.sql("SELECT count(*) AS rows FROM very_late").show()
spark.catalog.dropTempView("very_late")
print("exists after the drop :", spark.catalog.tableExists("very_late"))
# Registering a view reads nothing. It names a query plan, and the plan runs every time
# the name is queried. A view is not a cache.

worse = spark.sql("""
    SELECT carrier, round(avg(dep_delay), 2) AS avg_delay
    FROM flights
    GROUP BY carrier
    HAVING avg(dep_delay) > (SELECT avg(dep_delay) FROM flights)
    ORDER BY avg_delay DESC
""")

worse.show()
print("carriers above the overall average :", worse.count())

spark.sql("""
    SELECT round(avg(dep_delay), 3) AS overall_average FROM flights
""").show()
spark.sql("""
    SELECT round(avg(a), 3) AS average_of_averages FROM
    (SELECT avg(dep_delay) AS a FROM flights GROUP BY carrier)
""").show()
# The overall average weights every flight equally. The average of averages weights every
# carrier equally. Neither is wrong ; knowing which one you computed is the job.

by_city = spark.sql("""
    SELECT a.municipality AS city,
           count(*) AS flights,
           round(avg(f.dep_delay), 2) AS avg_delay
    FROM flights f
    JOIN airports a ON f.origin = a.iata_code
    GROUP BY a.municipality
    HAVING count(*) > 20000
    ORDER BY avg_delay DESC
""")

by_city.show(10, truncate=False)

by_city.explain()
# The airports table is small, so Spark broadcasts it : BroadcastHashJoin.

spark.sql("""
    WITH per_pair AS (
        SELECT origin, carrier,
               count(*) AS flights,
               round(avg(dep_delay), 2) AS avg_delay
        FROM flights
        WHERE dep_delay IS NOT NULL
        GROUP BY origin, carrier
        HAVING count(*) > 5000
    ),
    ranked AS (
        SELECT per_pair.*,
               rank() OVER (PARTITION BY origin ORDER BY avg_delay DESC) AS worst_rank
        FROM per_pair
    )
    SELECT origin, carrier, flights, avg_delay
    FROM ranked WHERE worst_rank = 1
    ORDER BY avg_delay DESC
    LIMIT 10
""").show()
# A GROUP BY collapses the rows, and the answer is a whole row : the worst carrier at each airport.

api = (flights
    .filter(F.col("dep_delay").isNotNull())
    .groupBy("carrier")
    .agg(F.count("*").alias("flights"),
         F.round(F.avg("dep_delay"), 2).alias("avg_delay"))
    .orderBy(F.desc("avg_delay")))

import re

api.explain()
by_carrier.explain()

def plan_of(df):
    return df._jdf.queryExecution().executedPlan().toString()

def normalised(plan):
    return re.sub(r"plan_id=\d+", "plan_id=x", re.sub(r"#\d+L?", "#x", plan))

print("same answer :", api.collect() == by_carrier.collect())
print("raw text equal :", plan_of(api) == plan_of(by_carrier))
print("same plan      :", normalised(plan_of(api)) == normalised(plan_of(by_carrier)))
# Spark numbers every expression and plan node as it builds them, so count#132L in one plan
# is count#135L in the other. The plans describe the same work : there is no performance
# argument either way.

table_name = "my_carrier_delays"

by_carrier.write.mode("overwrite").saveAsTable(table_name)
spark.sql(f"SELECT * FROM {table_name} ORDER BY avg_delay DESC LIMIT 5").show()

for t in spark.catalog.listTables():
    print(f"{t.name:22} {t.tableType}")

spark.sql(f"DROP TABLE IF EXISTS {table_name}")
print("dropped :", not spark.catalog.tableExists(table_name))
# A view is TEMPORARY and a saved table is MANAGED. The views go with the session ;
# only the table is still there for the next one.