class UsedFunctions {
  import scala.util.Random
  import scala.math._
  import org.apache.spark.sql.functions._
  def randomString(chars: String, length: Int): String = {
     (0 until length).map(_ => chars(Random.nextInt(chars.length))).mkString
  }
  def clustered(id : Int, numRows: Int) : Double  = { (id - 1).floor/numRows }
  def scattered(id : Int, numRows: Int) : Double  = { (id - 1 % numRows).abs }
  def randomised(seed: Int, numRows: Int) : Double  = { (Random.nextInt(seed) % numRows).abs }
  def padString(id: Int, chars: String, length: Int): String = {
      val n = (Math.log10(id)+1).toInt // How many digits id has
      (0 until length - n).map(_ => chars(Random.nextInt(chars.length))).mkString + id.toString
  }
  def padSingleChar(chars: String, length: Int): String = {
     (0 until length).map(_ => chars(Random.nextInt(chars.length))).mkString
  }
}

println ("\nStarted at"); spark.sql("SELECT FROM_unixtime(unix_timestamp(), 'dd/MM/yyyy HH:mm:ss.ss') ").collect.foreach(println)
  
//spark.udf.register("randomString", randomString(_:String, _:Int))
case class columns (
                       id: Int
                     , clustered: Double
                     , scattered: Double
                     , randomised: Double
                     , random_string: String
                     , small_vc: String
		     , padding: String
                   )

//val chars = ('a' to 'z') ++ ('A' to 'Z') ++ ('0' to '9') ++ ("-!£$")
val chars = ('a' to 'z') ++ ('A' to 'Z')

//
//get the max(ID) from fullyQualifiedTableName
//
val numRows = 50000   //do in increment of 50K rows otherwise you blow up driver memory!
//
// Check if table exist otherwise create it
//
val DB = "test"
val tableName = "randomData"
val fullyQualifiedTableName =  DB + "."+ tableName
var rows = 0
var sqltext  = ""
if (spark.sql(s"SHOW TABLES IN ${DB} like '${tableName}'").count() == 1) {
     rows = spark.sql(s"SELECT COUNT(1) FROM ${fullyQualifiedTableName}").collect.apply(0).getLong(0).toInt
 } else
 {   println (s"\n Table ${fullyQualifiedTableName} does not exist, creating table ")
   sqltext = s"""
     CREATE TABLE ${fullyQualifiedTableName}(
       ID INT
     , CLUSTERED INT
     , SCATTERED INT
     , RANDOMISED INT
     , RANDOM_STRING VARCHAR(50)
     , SMALL_VC VARCHAR(50)
     , PADDING  VARCHAR(4000)
    )
    STORED AS PARQUET
    """
    spark.sql(sqltext)
 }

var start = 0
if (rows == 0) {
    start = 1
} else {
   val maxID = spark.sql(s"SELECT MAX(id) FROM ${fullyQualifiedTableName}").collect.apply(0).getInt(0) 
   start = maxID + 1
}
val end = start + numRows - 1
println (" starting at ID = " + start + " , ending on = " +  end )
val usedFunctions = new UsedFunctions

val text = ( start to end ).map(i => 
             (
                 i.toString
               , usedFunctions.clustered(i,numRows).toString
               , usedFunctions.scattered(i,numRows).toString
               , usedFunctions.randomised(i,numRows).toString
               , usedFunctions.randomString(chars.mkString(""),50)
               , usedFunctions.padString(i, " ", 50)
               , usedFunctions.padSingleChar("x", 4000)
             )
           ).
    toArray
val df = sc.parallelize(text).
                              map(p => columns(
                                                  p._1.toString.toInt
                                                , p._2.toString.toDouble
                                                , p._3.toString.toDouble
                                                , p._4.toString.toDouble
                                                , p._5.toString
                                                , p._6.toString
                                                , p._7.toString
                                              )
                                 ).
    toDF
//
// register DF as tempTable
//
df.registerTempTable("tmp")
  //
  // Put data in Hive table. Clean up is already done
  //
  sqltext = s"""
  INSERT INTO TABLE ${fullyQualifiedTableName}
  SELECT
          ID
        , CLUSTERED
        , SCATTERED
        , RANDOMISED
        , RANDOM_STRING
        , SMALL_VC
        , PADDING
  FROM tmp
  """
  spark.sql(sqltext)
  spark.sql(s"""select MIN(id) AS minID, MAX(id) AS maxID from ${fullyQualifiedTableName}""").show
 spark.sql(s"select * from ${fullyQualifiedTableName} order by id").show(false)
  println ("\nFinished at"); spark.sql("SELECT FROM_unixtime(unix_timestamp(), 'dd/MM/yyyy HH:mm:ss.ss') ").collect.foreach(println)
  sys.exit()