Simple Calliope Save To Cassandra Test

CalliopeToCassandra.scala

def saveToCassandra(sc:SparkContext, total:Int): Long = {

    val cas = CasBuilder.cql3.withColumnFamily("keyspace", "raw").onHost(host).onPort(port)
      .saveWithQuery("UPDATE keyspace.raw set metric=?, vals=?")


    implicit def TsToCasKeys(src:Array[String]): Map[String, ByteBuffer] = src match {
       case Array(device_id, aggregation_type, t0, metric, vals) =>
         Map("device_id" -> device_id, "aggregation_type" -> aggregation_type, "t0" -> t0)
     }

    implicit def toToCassandraVals(src:Array[String]) : List[ByteBuffer] = src match {
        case Array(device_id, aggregation_type, t0, metric, vals) => List(metric, vals)
    }

    val entries = for (i <- 1 to total) yield {
      Array(s"devy$i", "aggr", "1000", "sum", (i to i+10).mkString(","))
    }

    val rdd = sc.parallelize(entries,8)

    rdd.cql3SaveToCassandra(cas)

    total

  }