使用spark-cassandra连接器的Cassandra插入器性能

Question

import java.util.UUID

import org.apache.spark.{SparkContext, SparkConf}
import org.joda.time.DateTime
import com.datastax.spark.connector._

case class TestEntity(id:UUID, category:String, name:String,value:Double, createDate:DateTime, tag:Long)

object SparkConnectorContext {
  val conf = new SparkConf(true).setMaster("local")
    .set("spark.cassandra.connection.host", "192.168.xxx.xxx")
  val sc = new SparkContext(conf)
}
object TestRepo {
  def insertList(list: List[TestEntity]) = {
    SparkConnectorContext.sc.parallelize(list).saveToCassandra("testKeySpace", "testColumnFamily")
  }
}
object TestApp extends App {
  val start = System.currentTimeMillis()
  TestRepo.insertList(Utility.generateRandomData())
  val end = System.currentTimeMillis()
  val timeDiff = end-start
  println("Difference (in millis)= "+timeDiff)
}

谁能告诉我为什么火花连接器要花这么多时间插入？我在代码中做了什么错误吗？或者使用spark-cassandra连接器进行插入操作是否不可取？

Answer 1

看起来您在计时中包含了并行化操作。另外，由于您的spark worker运行在与Cassandra不同的机器上，saveToCassandra操作将是网络上的写操作。

尝试将系统配置为在Cassandra节点上运行spark workers。然后在单独的步骤中创建一个RDD，并调用count（）这样的操作将数据加载到内存中。另外，您可能希望persist（）或cache（）RDD以确保它在内存中进行测试。

然后只对缓存的RDD的saveToCassandra计时。