spark_streaming_snippet.py

ds_sales = (
    df_sales.selectExpr("CAST(value AS STRING)")
    .select(F.from_json("value", schema=schema).alias("data"))
    .select("data.*")
    .withWatermark("transaction_time", "10 minutes")
    .groupBy("product_id", F.window("transaction_time", "10 minutes", "5 minutes"))
    .agg(F.sum("total_purchase"), F.sum("quantity"))
    .orderBy(F.col("window").desc(), F.col("sum(total_purchase)").desc())
    .select(
        "product_id",
        F.format_number("sum(total_purchase)", 2).alias("sales"),
        F.format_number("sum(quantity)", 0).alias("drinks"),
        "window.start",
        "window.end",
    )
    .coalesce(1)
    .writeStream.queryName("streaming_to_console")
    .trigger(processingTime="1 minute")
    .outputMode("complete")
    .format("console")
    .option("numRows", 10)
    .option("truncate", False)
    .start()
)

ds_sales.awaitTermination()