// scalastyle:off println

package main.scala.chapter2

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

/**
  * Użycie: MnMcount <mnm_file_dataset>
  */
object MnMcount {
  def main(args: Array[String]) {
    val spark = SparkSession
      .builder
      .appName("MnMCount")
      .getOrCreate()

    if (args.length < 1) {
      print("Usage: MnMcount <mnm_file_dataset>")
      sys.exit(1)
    }
    // Pobranie nazwy pliku zawierającego zbiór danych dotyczących cukierków M&M’s.
    val mnmFile = args(0)
    // Wczytanie pliku do egzemplarza DataFrame w Sparku.
    val mnmDF = spark.read.format("csv")
      .option("header", "true")
      .option("inferSchema", "true")
      .load(mnmFile)
    // Wyświetlenie egzemplarza DataFrame.
    mnmDF.show(5, false)

    // Agregowanie liczby wystąpień kolorów przez wywołanie groupBy() dla pól "State" i "Color".
    // Wywołanie orderBy() i ustawienie wartości w kolejności malejącej.
    val countMnMDF = mnmDF.select("State", "Color", "Count")
        .groupBy("State", "Color")
        .sum("Count")
        .orderBy(desc("sum(Count)"))

    // Wyświetlenie zagregowanych wyników dla wszystkich stanów i kolorów.
    countMnMDF.show(60)
    println(s"Total Rows = ${countMnMDF.count()}")
    println()

    // Zagregowane wyniki dla Kalifornii są wyszukiwane za pomocą filtrowania.
    val caCountMnNDF = mnmDF.select("*")
      .where(col("State") === "CA")
      .groupBy("State", "Color")
      .sum("Count")
      .orderBy(desc("sum(Count)"))

    // Wyświetlenie zagregowanych wyników dla Kalifornii.
    caCountMnNDF.show(10)
  }
}
// scalastyle:on println
