medium_ref_sstable_cql_writer.scala

def SSTableWriter(it: Iterator[Row], createTableStatement: String, insertStatement: String, cassHosts: String): Unit = {
  val uuid = UUID.randomUUID().toString
  val dir = "/tmp/" + uuid
  new File(dir).mkdirs()
  val writer = CQLSSTableWriter.builder().inDirectory(dir).forTable(createTableStatement)
    .using(insertStatement).withPartitioner(new Murmur3Partitioner()).build()

  while (it.hasNext) {
    val row = it.next()
    val rowValues = new java.util.ArrayList[AnyRef](row.length)
    row.schema.fields.indices.foreach(i => rowValues.add(row.get(i).asInstanceOf[AnyRef]))
    writer.addRow(rowValues)
  }
  writer.close()
  flushSSTableToCassandra(cassHosts, dir)
}