DataFrame 写入速度比 RDD 慢 10 倍，在 spark 中保存到 cassandra

Question

我尝试将 cassandra 保存的表与 RDD 和 DataSet 的 30,000 条记录进行比较。我发现数据集保存速度比 RDD 慢 10 倍。 该表有 4 个分区键。

 DSE Version :5.1.7
 Spark version :2.0.1
 Nodes:6( 20 cores each 6g)
 Using Spark Standalone

我们使用了以下 spark 配置：

1. spark.scheduler.listenerbus.eventqueue.size=100000
2. spark.locality.wait=1
3. spark.dse.continuous_paging_enabled=false
4. spark.cassandra.input.fetch.size_in_rows=500
5. spark.cassandra.connection.keep_alive_ms=10000
6. spark.cassandra.output.concurrent.writes=2000
7. num-cpu-cores=48
8. 每节点内存=3g
9. spark.executor.cores=3
10. spark.cassandra.output.ignoreNulls=true
11. spark.cassandra.output.throughput_mb_per_sec=10
12. spark.serializer=org.apache.spark.serializer.KryoSerializer
13. spark.cassandra.connection.local_dc=dc1
14. spark.cassandra.connection.compression=LZ4
15. spark.cassandra.connection.connections_per_executor_max=20

以下是相同的示例代码：

val sparkSession = SparkSession.builder().config(conf).getOrCreate()

import sparkSession.implicits._

val RDD1 = sc.cassandraTable[TableName]("keySpace1", "TableName")
           .where("id =?,id)

RDD1.saveToCassandra("keySpace1", "TableName")

var DS1 = sparkSession.read
           .format("org.apache.spark.sql.cassandra")
           .options(Map("table" -> "TableName", "keyspace" ->"keySpace1"))
           .load()
           .where("id ='"+ id +"'").as[CaseClassModel]

DS1.write.format("org.apache.spark.sql.cassandra")          
  .mode(SaveMode.Append).option("table", "TableName1")                
  .option("keyspace", "KeySpace1")
  .save()

Answer 1

由于 Dataframe 和 RDD 方法都使用相同的底层保存代码，除非转换为 DF 类型的开销非常高，否则您不太可能看到如此巨大的差异。在我们自己对数十亿行的测试中，我们发现速度只有百分之几的差异。

虽然 30k 条记录是一个非常小的数量，并且任何数量的开销都可能变得相关，但我认为最可能的原因是 where 子句中的查找在 RDD 和 DF 代码中的解释不同。我会检查以确保它被正确推送到 DF 代码中（请参阅explain 了解 DF 负载。）