%run "./Stream-Generator"

%scala

DummyDataGenerator.clean()

DummyDataGenerator.start(5)

class DummyDataGenerator:

  streamDirectory = "dbfs:/tmp/{}/new-purchases".format("EMAIL") # Whatever e-mail you used to sign up for databricks

None # suppress output

%scala

import scala.util.Random

import java.io._

import java.time._

// http://spark.apache.org/docs/latest/sql-programming-guide.html#other-configuration-options

// https://stackoverflow.com/questions/33297689/number-reduce-tasks-spark

spark.conf.set("spark.sql.shuffle.partitions", 200)

// Make the username available to all other languages.

val username = com.databricks.logging.AttributionContext.current.tags(com.databricks.logging.BaseTagDefinitions.TAG_USER);

spark.conf.set("com.databricks.training.username", username)

object DummyDataGenerator extends Runnable {

  var runner : Thread = null;

  val className = getClass().getName()

  val streamDirectory = s"dbfs:/tmp/$username/new-purchases"

  val rand = new Random(System.currentTimeMillis())

  var maxDuration = 3 * 60 * 1000 // default to a couple of minutes

  def clean() {

    System.out.println("Removing old files for dummy data generator.")

    dbutils.fs.rm(streamDirectory, true)

    if (dbutils.fs.mkdirs(streamDirectory) == false) {

      throw new RuntimeException("Unable to create temp directory.")

    }

  }

  def run() {

    val date = LocalDate.now()

    val start = System.currentTimeMillis()

    while (System.currentTimeMillis() - start < maxDuration) {

      try {

        val dir = s"/dbfs/tmp/$username/new-purchases"

        val tempFile = File.createTempFile("purchases-", "", new File(dir)).getAbsolutePath()+".csv"

        val writer = new PrintWriter(tempFile)

        val productId = rand.nextInt(5)+1

        val purchaseTime = LocalDateTime.now().plusHours(-4)

        val revenue = rand.nextInt(1000)+1

        println(s"- Product #$productId sold at $purchaseTime for $revenue dollars")

        writer.println(s""" "$productId","$purchaseTime","$revenue" """.trim)

        writer.close()

        // wait a couple of seconds

        Thread.sleep(rand.nextInt(5000))

      } catch {

        case e: Exception => {

          printf("* Processing failure: %s%n", e.getMessage())

          return;

        }

      }

    }

    println("No more purchases!")

  }

  def start(minutes:Int = 5) {

    maxDuration = minutes * 60 * 1000

    if (runner != null) {

      println("Stopping dummy data generator.")

      runner.interrupt();

      runner.join();

    }

    println(s"Running dummy data generator for $minutes minutes.")

    runner = new Thread(this);

    runner.start();

  }

  def stop() {

    start(0)

  }

}

DummyDataGenerator.clean()

displayHTML("Imported streaming logic...") // suppress output

%run "./Stream-Generator"

%scala

// Clean any temp files from previous runs.

DummyDataGenerator.clean()

// Generate data for 5 minutes.

// To force it to stop rerun with 0.

DummyDataGenerator.start(5)

from pyspark.sql.functions import col, date_format, unix_timestamp, window

from pyspark.sql.types import StructType

spark.conf.set("spark.sql.shuffle.partitions", "8")

purchaseSchema = (StructType()

  .add("ProductId", "integer")

  .add("PurchaseTime", "string")                  

  .add("Revenue", "integer")

)

streamingDF = (spark.readStream

  .schema(purchaseSchema)

  .csv(DummyDataGenerator.streamDirectory)

  .withColumn("PurchaseTime", unix_timestamp("PurchaseTime", "yyyy-MM-dd'T'HH:mm:ss").cast("timestamp"))

  .withWatermark("PurchaseTime", "5 minute")

  .groupBy(window("PurchaseTime", "1 seconds"), "ProductId")

  .sum("Revenue")

  .select(col("window.start").alias("Start"), "ProductId", col("sum(Revenue)").alias("Revenue"))

  .orderBy("start", "ProductId")

  .select(date_format("start", "HH:mm:ss").alias("Time"), "ProductId", "Revenue")

)

display(streamingDF)