untitled

import org.apache.spark.sql.{SparkSession, SaveMode}
import org.apache.spark.sql.functions._
import java.text.SimpleDateFormat

object DailyRevenueJob {

  case class Order(orderId: String, customerId: String, amount: Double,
                   currency: String, createdAt: String)

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("DailyRevenueJob")
      .master("local[*]")
      .config("spark.sql.shuffle.partitions", "2000")
      .getOrCreate()
    import spark.implicits._

    val orders = spark.read
      .option("inferSchema", "true")
      .option("header", "true")
      .csv("s3://prod-data/orders/")

    val customers = spark.read.parquet("s3://prod-data/customers/")

    // Normalise the date
    val toDay = udf((ts: String) => {
      val in  = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
      val out = new SimpleDateFormat("yyyy-MM-dd")
      out.format(in.parse(ts))
    })

    val withDay = orders.withColumn("day", toDay($"createdAt"))
    withDay.cache()

    println(s"Total orders: ${withDay.count()}")

    // Enrich with customer info
    val enriched = withDay.join(customers, withDay("customerId") === customers("id"))

    // Revenue per customer per day
    val revenue = enriched
      .as[Order]
      .rdd
      .map(o => ((o.customerId, o.createdAt.take(10)), o.amount))
      .groupByKey()
      .mapValues(_.sum)
      .collect()

    var grandTotal = 0.0
    revenue.foreach { case (_, amt) => grandTotal += amt }
    println(s"Grand total revenue: $grandTotal")

    // Persist the result
    revenue.toSeq
      .toDF("key", "revenue")
      .repartition(1)
      .write
      .mode(SaveMode.Overwrite)
      .parquet("s3://prod-data/reports/daily_revenue/")

    spark.stop()
  }
}