C*迁移-将1B+行表数据移动到新的模式表

Question

我使用DSE 4.7 datastax-enterprise，C* 2.1.5，spark 1.2.1，并需要将数据从一个大表迁移到具有不同模式和额外列的新的空表，这些数据需要从大表中的一个删除列生成。

我知道，将表数据迁移到另一个具有新模式的表可以通过spark或将命令复制到cqlsh中的csv文件来完成，但我对一个工具很感兴趣，它可以为将来的迁移提供长期解决方案，以及更多的管理和规划迁移的选项。

我认为这是一个常见的问题，我没有找到任何固着的解决办法。

有什么想法吗？

Answer 1

我确信火花是这项工作的最佳工具。我已经测试了下面的代码，结果很好。

import java.sql.Date
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.{Row, SQLContext}
import com.datastax.spark.connector._
import com.datastax.spark.connector.cql.CassandraConnector
import java.sql._
import com.github.nscala_time.time.Imports._


object Migration {
  def main(args: scala.Array[String]) {

    def changeDate(created: java.util.Date) : String = {
        var sDate = new DateTime(created)
        var sDay = sDate.getDayOfMonth()
        var sMonth = sDate.getMonthOfYear()
        var sYear = sDate.getYear()
        var created_date = "" + sYear + "-" + sMonth + "-" + sDay
        created_date //return
    }

    //spark configuration
    val conf = new SparkConf().setAppName("migration")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val connector = CassandraConnector(conf)

    val rdd = sc.cassandraTable("keyspace", "table_a")

    println("Starting migration...")

    rdd.map(row => { 
        val x = new java.util.Date(row.getLong("x"))
        val y = new java.util.Date(row.getLong("y"))
        val z = row.getString("z")
        val t = row.getString("t")
        val k = changeDate(x)

        connector.withSessionDo(session => {
            val statement = session.prepare(s"INSERT INTO keyspace.table_b (k, y, z, x, t) " + "values (?, ?, ?, ?, ?)")
            val bound = statement.bind(k, y, z, x, t)
            session.executeAsync(bound)
        })
    }).foreach(x => x.getUninterruptibly())

    println("Done.")



 } }