clicks.scala

import java.util.concurrent.TimeUnit

import monix.execution.Scheduler.{global => scheduler}
import org.apache.spark.sql.{Encoders, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{StringType, StructField, StructType}

object Test extends App {


  val spark = SparkSession.builder().appName("test").master("local[5]").getOrCreate()

  import spark.implicits._

  val stream = spark.readStream
    .format("json")
    .option("path", "/opt/test/events/*")
    .schema(Encoders.product[Test].schema).load() 
  stream.createOrReplaceTempView("clicks_stream")
  
  val userIdMapDf = spark.read
    .format("com.databricks.spark.csv")
    .option("delimiter", ",")
    .schema(idSchema)
    .load("/opt/test/dw/users/cid/2018/02/13/*")
  val cidDf = userIdMapDf.withColumn("old_cid", when($"did".isNull, $"aid").otherwise($"did")).drop("did", "aid")
  cidDf.createOrReplaceTempView("users")

  spark.sql("SELECT * FROM clicks_stream LEFT JOIN users ON clicks_stream.user_cid = users.old_cid")
    .withColumn("is_new", when($"old_cid".isNull, true).otherwise(false)).createOrReplaceTempView("clicks")


  val c = scheduler.scheduleWithFixedDelay(
    0, 1, TimeUnit.SECONDS,
    new Runnable {
      def run(): Unit = {
        val userIdMapDf = spark.read
          .format("com.databricks.spark.csv")
          .option("delimiter", ",")
          .schema(idSchema)
          .load("/opt/test/dw/users/cid/2018/02/13/*")
        val cidDf = userIdMapDf.withColumn("old_cid", when($"did".isNull, $"aid").otherwise($"did")).drop("did", "aid")
        cidDf.createOrReplaceTempView("users")

        spark.sql("SELECT * FROM clicks_stream LEFT JOIN users ON clicks_stream.user_cid = users.old_cid")
        .withColumn("is_new", when($"old_cid".isNull, true).otherwise(false)).createOrReplaceTempView("clicks")
       
      }
    })


  var outStream = spark.sql("SELECT user_cid, old_cid, event_name, is_new FROM clicks")
    .writeStream
    .outputMode("append")
    .format("console")
    .start()

  outStream.awaitTermination()
}