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

Posted 2023-04-17

技术标签:

【中文标题】Spark-Cassandra 写入所需的时间比预期的要长【英文标题】：Spark-Cassandra write takes longer than expected 【发布时间】：2017-01-30 18:15:47 【问题描述】：

我有一个 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"))

对于大约一百万条记录，STORE 1 需要将近 40 秒，而 STORE 2（对 rdd3 的轻微修改）需要一分钟以上。我不确定我哪里出错了，或者为什么要花这么多时间。我的spark环境如下：

具有 6 个节点的 DSE 4.8.9 70 GB 内存每个 12 个核心

感谢任何帮助。

【问题讨论】：

您是否尝试过将 rdd3 添加到检查点并看到它变得更快？不。它不会走得更快 【参考方案1】：

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

joinWithCassandra — 随机 C* 读取 saveToCassandra — 秒 C* 写入火花重新分区？ /减少

（我预计 saveToCassadndra 需要一半的时间）如果您在需要负 12-20 秒的时间让 spark 启动执行程序和其他事情之前不运行任何查询

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

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

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

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