PySpark DataFrame Basics in 10 Operations

from pyspark.sql import SparkSession, functions as F, Window
 
spark = SparkSession.builder.appName("orders").getOrCreate()

orders = spark.read.parquet("s3://lake/raw/orders/")
paid = orders.select("order_id", "store_id", "amount", "status", "ts").filter(F.col("status") == "paid")

paid = paid.withColumn("business_date", F.to_date("ts"))

stores = spark.read.parquet("s3://lake/dim/stores/")
enriched = paid.join(stores, "store_id", "left")

daily = enriched.groupBy("business_date", "region").agg(
    F.sum("amount").alias("net_sales"),
    F.countDistinct("order_id").alias("orders"),
)

w = Window.partitionBy("region").orderBy("business_date")
daily = daily.withColumn("running_sales", F.sum("net_sales").over(w))

w = Window.partitionBy("order_id").orderBy(F.col("ts").desc())
latest = orders.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")

both = jan.unionByName(feb, allowMissingColumns=True)   # by name, not position

daily.write.mode("overwrite").partitionBy("business_date").parquet("s3://lake/curated/daily/")