package hozza.network

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.springframework.beans.factory.annotation.Value
import org.springframework.scheduling.annotation.Async
import org.springframework.stereotype.Service

@Service
class Test
{
  @Value("${spark.datasource.url}")
  val url: String = null

  @Value("${spring.datasource.username}")
  val username: String = null

  @Value("${spring.datasource.password}")
  val password: String = null

  val driver: String = "jdbc"

  @Async
  def test() =
  {
    val conf = new SparkConf().setAppName("hozza.network").setMaster("local")
    val spark = SparkSession.builder().config(conf).getOrCreate()
    val df = spark.read.format(driver)
      .option("url", url)
      .option("user", username)
      .option("password", password)

    /*df.option("dbtable", "country")
      .load()
      .createTempView("country")

    df.option("dbtable", "streamer")
      .load()
      .createTempView("streamer")

    df.option("dbtable", "person")
      .option("fetchSize", 10000)
      .option("partitionColumn", "id")
      .option("lowerBound", 3)
      .option("upperBound", 44305655)
      .load()
      .createTempView("person")*/

    df.option("dbtable", "watching")
      .option("fetchSize", 10000)
      .option("partitionColumn", "date_time_watched")
      .option("numPartitions", 100)
      .option("lowerBound", "2021-10-01")
      .option("upperBound", "2022-04-30")
      .option("dateFormat", "yyyy-mm-dd")
      .load()
      .createTempView("watching")

    spark.sql(" SELECT k1.streamer as source, k2.streamer as target, count(k1.person) as weight FROM watching AS k1 INNER JOIN watching AS k2 ON k1.streamer != k2.streamer AND k1.person = k2.person WHERE k2.streamer > k1.streamer GROUP BY k1.streamer,k2.streamer ORDER BY NULL").collect().foreach(println)
  }
}
