Spark Cassandra连接器的正确使用

Question

class CassandraStorage(conf:SparkConf) {
  val session = CassandraConnector(conf).openSession()
  def store (t:Thingy) : Unit = {
    //session.execute cql goes here
  }
}

这是个好办法吗？我需要担心关闭会话吗？我在哪里/怎么做最好？任何指针都很感激。

Answer 1

您确实希望使用withSessionDo，因为它实际上不会在每次访问时打开和关闭会话。在下面，withSessionDo访问JVM级别的会话。这意味着每个集群配置每个节点只有一个会话对象。

这意味着代码

val connector = CassandraConnector(sc.getConf)
sc.parallelize(1 to 10000000L).map(connector.withSessionDo( Session => stuff)

无论每台机器有多少核，在每个执行器JVM上只会生成一个集群和会话对象。

sc.parallelize(1 to 10000000L)
  .mapPartitions(it => connector.withSessionDo( session => 
      it.map( row => do stuff here )))