Spark Cassandra写入所需时间比预期长

Question

我有一个spark作业，它从一个cassandra表中读取数据，并将结果转储回两个表，只需稍作修改。我的问题是这项工作比预期的要长得多。

代码如下：

val range = sc.parallelize(0 to 100)

val rdd1 = range.map(x => (some_value, x)).joinWithCassandraTable[Event](keyspace_name, table2).select("col1", "col2", "col3", "col4", "col5", "col6", "col7").map(x => x._2)

val rdd2: RDD[((Int, String, String, String), Iterable[Event])] = rdd1.keyBy(r => (r.col1, r.col2, r.col3, r.col4 )).groupByKey

val rdd3 = rdd2.mapValues(iter => someFunction(iter.toList.sorted))

//STORE 1

rdd3.map(r => (r._1._1, r._1._2, r._1._3, r._1._4, r._2.split('|')(1).toDouble )).saveToCassandra(keyspace_name, table1, SomeColumns("col1","col2", "col3","col4", "col5"))

//STORE 2  

rdd3.map(r => (to, r._1%100, to, "MANUAL_"+r._1+"_"+r._2+"_"+r._3+"_"+r._4+"_"+java.util.UUID.randomUUID(), "M", to, r._4, r._3, r._1, r._5, r._2) ).saveToCassandra(keyspace_name, table2, SomeColumns("col1", "col2", "col3", "col4", "col5", "col6", "col7", "col8", "col9", "col10", "col11"))

对于大约一百万条记录，STORE1需要将近40秒，STORE2（对rdd3稍作修改）需要一分钟以上。不确定我哪里出错了，也不知道为什么要花这么多时间。我的火花环境如下：

DSE 4.8.9,6个节点70 GB RAM，每个12个内核

任何帮助，不胜感激。

Answer 1

让我做我的猜测。需要日志、性能监控输出和 C* 数据模型才能获得更精确的答案。但是一些数学：你有

• 加入Cassandra-随机C*读取
• SaveToCassandra-秒C*写入
• 火花重新分区？/减少

（我预计SaveToCassadndra需要一半的时间）如果你不运行任何查询之前，你需要减去12-20秒的火花启动执行器和其他东西

因此，对于6个节点上的1M条目，您可以得到40秒：
 1000000/6/40=4166记录/秒/节点。这还不错。具有混合工作负载的每个节点10K/s是一个很好的结果。

第二次写入是第二次的2倍（11列，5列），它在第一次写入之后运行，所以我希望Cassandra此时开始将之前的数据溢出到磁盘，这样您可以在这里获得更多的性能降级。

我是否正确理解当您添加 rdd3.cache（） 调用时，第二次运行没有任何变化？太奇怪了。

是的，您可以通过调整C*数据模型和Spark/C*参数获得更好的结果