import scala.util.Random
import scala.util.Random._

// make a sample to produce a mixture of two normal RVs with standard deviation 1 but with different location or mean parameters
def myMixtureOf2Normals( normalLocation: Double, abnormalLocation: Double, normalWeight: Double, r: Random) : (String, Double) = {
  val sample = if (r.nextDouble <= normalWeight) {r.nextGaussian+normalLocation } 
               else {r.nextGaussian + abnormalLocation} 
  Thread.sleep(5L) // sleep 5 milliseconds
  val now = (new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")).format(new java.util.Date())
  return (now,sample)
   }

import scala.util.Random
import scala.util.Random._
myMixtureOf2Normals: (normalLocation: Double, abnormalLocation: Double, normalWeight: Double, r: scala.util.Random)(String, Double)

val r = new Random(1L)
println(myMixtureOf2Normals(1.0, 10.0, 0.99, r), myMixtureOf2Normals(1.0, 10.0, 0.99, r))
// should always produce samples as (0.5876430182311466,-0.34037937678788865) when seed = 1L

((2020-11-16 10:47:55.774,0.5876430182311466),(2020-11-16 10:47:55.780,-0.34037937678788865))
r: scala.util.Random = scala.util.Random@cb2fa8b

display(sc.parallelize(Vector.fill(1000){myMixtureOf2Normals(1.0, 10.0, 0.99, r)}).toDF.select("_2")) // histogram of 1000 samples

dbutils.fs.rm("/datasets/streamingFilesNormalMixture/",true) // this is to delete the directory before staring a job

res2: Boolean = false

val r = new Random(12345L) // set seed for reproducibility
var a = 0;
// for loop execution to write files to distributed fs
for( a <- 1 to 5){
  // make a DataSet
  val data = sc.parallelize(Vector.fill(100){myMixtureOf2Normals(1.0, 10.0, 0.99, r)}) // 100 samples from mixture
               .coalesce(1) // this is to make sure that we have only one partition per dir
               .toDF.as[(String,Double)]
  val minute = (new java.text.SimpleDateFormat("mm")).format(new java.util.Date())
  val second = (new java.text.SimpleDateFormat("ss")).format(new java.util.Date())
  // write to dbfs
  data.write.mode(SaveMode.Overwrite).csv("/datasets/streamingFilesNormalMixture/" + minute +"_" + second)
  Thread.sleep(5000L) // sleep 5 seconds
}

r: scala.util.Random = scala.util.Random@5704bfda
a: Int = 0

display(dbutils.fs.ls("/datasets/streamingFilesNormalMixture/"))

display(dbutils.fs.ls("/datasets/streamingFilesNormalMixture/48_43/"))

val df_csv = spark.read.option("inferSchema", "true").csv("/datasets/streamingFilesNormalMixture/48_43/*.csv")

df_csv: org.apache.spark.sql.DataFrame = [_c0: string, _c1: double]

df_csv.count() // 100 samples per file

res8: Long = 100

df_csv.show(10,false) // first 10

+-----------------------+--------------------+
|_c0                    |_c1                 |
+-----------------------+--------------------+
|2020-11-16 10:48:42.690|2.0531657985840983  |
|2020-11-16 10:48:42.696|1.7928797637680196  |
|2020-11-16 10:48:42.701|2.9329556976986013  |
|2020-11-16 10:48:42.706|1.1087520027663345  |
|2020-11-16 10:48:42.711|1.2115868818351045  |
|2020-11-16 10:48:42.716|1.9163661519192294  |
|2020-11-16 10:48:42.722|1.6917128257752045  |
|2020-11-16 10:48:42.727|1.0095879056962782  |
|2020-11-16 10:48:42.732|-0.13611276130309613|
|2020-11-16 10:48:42.737|2.2939319088848023  |
+-----------------------+--------------------+
only showing top 10 rows