Spark和Cassandra并行处理

Question

假设数组如下所示：

值ips=数组（1,2,3,4,5)

数组中最多可以有100.000个值。

ips.foreach(ip =>{
- read data from Casandra for specific "ip" // for each IP there is different amount of data to read (within the functions I determine start and end date for each IP)
- process it
- save it back to Cassandra})

另一方面，如果我这样做：

val IPRdd = sc.parallelize(Array(1,2,3,4,5))
IPRdd.foreach(ip => {
- read data from Cassandra // I need to use spark context to make the query
-process it
save it back to Cassandra})

我得到serialization异常，因为spark正在尝试序列化spark上下文，而spark上下文是不可序列化的。

如何使这个工作，但仍然利用并行性。

这是我得到的咒语：

Answer 1

最简单的方法是使用Spark Cassandra连接器，它可以处理连接池和序列化。

有了它，你就可以做一些事情，比如

sc.parallelize(inputData, numTasks)
  .mapPartitions {  it =>
    val con = CassandraConnection(yourConf)
    con.withSessionDo{ session =>
      //Use the session
    }
    //Do any other processing
  }.saveToCassandra("ks","table"

这将是完全手动操作的卡桑德拉连接。所有会话都将被自动合并和缓存，如果您准备了一条语句，这些会话也将被缓存在执行器上。

sc.parallelize(inputData, numTasks)
  .joinWithCassandraTable("ks","table") //Retrieves all records for which input data is the primary key
  .map( //manipulate returned results if needed )
  .saveToCassandra("ks","table")