从spark集群向cassandra集群写入dataframe：分区和性能调优

Question

我有两个集群-1。Cloudera Hadoop-Spark作业在这里运行2。云-卡桑德拉星团，多DC

在编写从spark作业到cassandra集群的dataframe时，我在编写之前在spark中进行了重新分区(repartioncount=10)。见下文：

import org.apache.spark.sql.cassandra._
records.repartition(repartitionCount).write.cassandraFormat(table, keySpace)
  .mode(SaveMode.Append)
  .options(options)
  .option(CassandraConnectorConf.LocalDCParam.name, cassandraDC.name)
  .option(CassandraConnectorConf.ConnectionHostParam.name, cassandraDC.hosts)
  .save()

在我的多租户spark集群中，对于一个有20M记录的spark批加载，以及以下配置，我看到了很多任务失败、资源抢占和动态失败。

spark.cassandra.output.batch.grouping.buffer.size=1000
spark.cassandra.output.batch.grouping.key=partition
spark.cassandra.output.concurrent.writes=20 
spark.cassandra.connection.compression=LZ4

PS：我一开始的理解是：对于20m行的负载，“重新分区”应该将负载均匀地分配到执行器上（每个分区有2m行），批处理将在这些分区级别（2m行）上进行。但是现在，我怀疑这是否会导致不必要的洗牌，如果spark-cassandra-connector在整个数据流级别（整个20m行）上进行批处理的话。

更新：删除“重新分区”大大降低了cloudera spark集群的性能（spark级别的默认分区设置是-spark.sql.shuffle.partitions:200)，所以我深入研究了一下，发现我最初的理解是正确的。请注意我的spark和cassandra集群是不同的。Datastax spark-cassandra-connector使用cassandra协调器节点打开每个分区的一个连接，所以我决定让它相同。正如Alex所建议的，我已经减少了并发写，我相信这会有所帮助。

Answer 1

您不需要在Spark中进行重新分区--只需将数据从Spark写入Cassandra即可，不要试图更改Spark Cassandra连接器的默认值--它们在大多数情况下都能正常工作。您需要查看发生了什么样的阶段故障--很可能只是因为spark.Cassandra.output.concurrent.writes=20（使用默认值(5))而重载了Cassandra--有时由于不重载Cassandra，作业也不会重新启动，所以编写器较少，有助于更快地写入数据。

附言。Spark.Cassandra.output.batch.grouping.key中的分区-它不是Spark分区，而是依赖于分区键列值的Cassandra分区。