spark_batch_snippet.py

ds_sales = (
    df_sales.selectExpr("CAST(value AS STRING)")
    .select(F.from_json("value", schema=schema).alias("data"))
    .select("data.*")
    .withColumn("row", F.row_number().over(window))
    .withColumn("quantity", F.sum(F.col("quantity")).over(window_agg))
    .withColumn("sales", F.sum(F.col("total_purchase")).over(window_agg))
    .filter(F.col("row") == 1)
    .drop("row")
    .select(
        "product_id",
        F.format_number("sales", 2).alias("sales"),
        F.format_number("quantity", 0).alias("quantity"),
    )
    .coalesce(1)
    .orderBy(F.regexp_replace("sales", ",", "").cast("float"), ascending=False)
    .write.format("console")
    .option("numRows", 25)
    .option("truncate", False)
    .save()
)