package main.scala.chapter3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions.{col, expr}

object Example3_7 {
  def main(args: Array[String]) {

    val spark = SparkSession
      .builder
      .appName("Example-3_7")
      .getOrCreate()

    if (args.length <= 0) {
      println("usage Example3_7 <file path to blogs.json")
      System.exit(1)
    }
    // Pobranie ścieżki dostępu do pliku JSON.
    val jsonFile = args(0)
    // Zdefiniowanie naszego schematu w sposób programistyczny.
    val schema = StructType(Array(StructField("Id", IntegerType, false),
      StructField("First", StringType, false),
      StructField("Last", StringType, false),
      StructField("Url", StringType, false),
      StructField("Published", StringType, false),
      StructField("Hits", IntegerType, false),
      StructField("Campaigns", ArrayType(StringType), false)))

    // Utworzenie egzemplarza DataFrame poprzez odczyt danych z pliku JSON,
    // z wykorzystaniem predefiniowanego schematu.
    val blogsDF = spark.read.schema(schema).json(jsonFile)
    // Wyświetlenie danych wyjściowych w postaci schematu egzemplarza DataFrame.
    blogsDF.show(truncate = false)
    // Wyświetlenie schematu.
    println(blogsDF.printSchema)
    println(blogsDF.schema)
    // Wyświetlenie kolumn i wyrażeń.
    blogsDF.select(expr("Hits") * 2).show(2)
    blogsDF.select(col("Hits") * 2).show(2)
    blogsDF.select(expr("Hits * 2")).show(2)
   // Wyświetlenie najpopularniejszych autorów.
   blogsDF.withColumn("Big Hitters", (expr("Hits > 10000"))).show()

  }
}
