Smart Thermostat AWS Spark - part Spark cluster retrieve temparature scenarios

smarttherm_part_cluster_retrieve_scenarios.scala

//spark settings
val sparkconf = new SparkConf().setAppName("tempcluster")
val sc = new SparkContext(sparkconf)

//scan hbase used scenario table to collect used scenarios that have been scored
val confonoff = HBaseConfiguration.create()
val scanonoff = new Scan()
scanonoff.addColumn("fd".getBytes(), "outtempdif".getBytes())
scanonoff.addColumn("fd".getBytes(), "tempdif".getBytes())
scanonoff.addColumn("fd".getBytes(), "score".getBytes())
scanonoff.addColumn("fd".getBytes(), "scenariokey".getBytes())
scanonoff.setStopRow("40b5af01_9999999999_40b5af01_9999999999_9999999".getBytes())
confonoff.set(TableInputFormat.INPUT_TABLE, "husedscenario")
confonoff.set(TableInputFormat.SCAN, convertScanToString(scanonoff))

//create RDD with used scenarios
val RDDonOff = sc.newAPIHadoopRDD(confonoff, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],classOf[org.apache.hadoop.hbase.client.Result])
val valRowsOnOff = RDDonOff.map(tuple => tuple._2
	).filter(result =>  Bytes.toString(result.getValue("fd".getBytes(), "score".getBytes())) != null //only select scored scenarios
	).map(result =>  ( Bytes.toString(result.getValue("fd".getBytes(), "scenariokey".getBytes())), (  Bytes.toString(result.getValue("fd".getBytes(), "outtempdif".getBytes())).toInt , Bytes.toString(result.getValue("fd".getBytes(), "tempdif".getBytes())).toInt    , Bytes.toString(result.getValue("fd".getBytes(), "score".getBytes())).toLong  )  )
	)
valRowsOnOff.cache()