scala – 如何在阅读spark中的cassandra分区时获得良好的性能？

我正在使用cassandra-connector将数据从cassandra分区读取到spark.我尝试了以下解决方案来读取partitions.I尝试通过尽可能多地创建rdds来并行化任务,但解决方案ONE和解决方案TWO都具有相同的性能.

在解决方案ONE中,我可以立即看到spark UI中的各个阶段.我试图在解决方案TWO中避免一个for循环.

在解决方案TWO中,阶段在相当长的时间之后出现.随着用户ID的数量增加,在阶段出现在解决方案TWO的火花UI中之前,时间显着增加.

Version
 spark - 1.1
 Dse - 4.6
 cassandra-connector -1.1

Setup
 3 - Nodes with spark cassandra
 Each node has 1 core dedicated to this task.
 512MB ram for the executor memory.

我的cassandra表架构,

CREATE TABLE   test (
   user text,userid bigint,period timestamp,ip text,data blob,PRIMARY KEY((user,userid,period),ip)
   );

第一解决方案

val users = List("u1","u2","u3")
 val period = List("2000-05-01","2000-05-01")
 val partitions = users.flatMap(x => period.map(y => (x,y))))
 val userids = 1 to 10
 for (userid <- userids){
 val rdds = partitions.map(x => sc.cassandraTable("test_keyspace","table1")
                                .select("data")
                                .where("user=?",x._1)
                                .where("period=?",x._2)
                                .where("userid=?,userid)
                          )
 val combinedRdd = sc.union(rdds)
 val result = combinedRdd.map(getDataFromColumns)
                    .coalesce(4)
                    .reduceByKey((x,y) => x+y)
                    .collect()
 result.foreach(prinltn)
 }

二解决方案：

val users = List("u1","2000-05-01")
 val userids = 1 to 10
 val partitions = users.flatMap(x => period.flatMap(
                  y => userids.map(z => (x,y,z))))

 val rdds = partitions.map(x => sc.cassandraTable("test_keyspace",x._3)
                     )
 val combinedRdd = sc.union(rdds)
 val result = combinedRdd.map(getDataFromColumns)
                    .coalesce(4)
                    .reduceByKey((x,y) => x+y)
                    .collect()
 result.foreach(prinltn)

为什么解决方案TWO不比解决方案ONE快？

我的理解是,由于所有分区都在一段查询,而数据分布在节点上,因此应该更快.
如果我错了,请纠正我.

sc.parallelize(partitions).joinWithCassandraTable("keyspace","table")

val cc = CassandraConnector(sc.getConf)
partitions.mapPartitions{ it => 
  cc.withSessionDo{ session =>
    session.execute( Some query )
  }
}