Spark-Cassandra：repartitionByCassandraReplica操作后分区数异常咨询

环境信息 #

• 集群：16节点，每个节点同时部署Spark与Cassandra
• Cassandra复制因子：3
• Spark配置：spark.sql.shuffle.partitions=96
• 使用连接器版本：Spark-Cassandra Connector 3.0.0

问题场景 #

我需要通过分区键将Spark Dataset与Cassandra表做关联，为了实现计算本地化，尝试使用repartitionByCassandraReplica方法，但该方法仅支持RDD API。于是我将Dataset转换为JavaRDD，执行repartitionByCassandraReplica后再转回Dataset，随后与Cassandra表执行Direct Join。但过程中出现两个异常：

1. 分区数不符合预期：初始Dataset有4个分区，执行repartitionByCassandraReplica后变为64个
2. 数据拉取节点错误：实际拉取数据的节点并非通过nodetool getendpoints确认的分区键存储节点

测试背景 #

本次针对4个分区键操作，每个分区键对应数千行数据，已通过nodetool getendpoints明确这些分区键的存储节点。

相关代码 #

// experimentlist是一个String列表，先转成Dataset，再转JavaRDD，执行repartitionByCassandraReplica后转回Dataset
// 要关联的Cassandra表名为experiment

List<ExperimentForm> tempexplist = experimentlist.stream()
            .map(s -> { ExperimentForm p = new ExperimentForm(); p.setExperimentid(s); return p; })
            .collect(Collectors.toList());
        
Encoder<ExperimentForm> ExpEncoder = Encoders.bean(ExperimentForm.class);
Dataset<ExperimentForm> dfexplistoriginal = sp.createDataset(tempexplist, Encoders.bean(ExperimentForm.class));
// 打印：DATASET: PartNum 4
System.out.println("DATASET: PartNum "+dfexplistoriginal.rdd().getNumPartitions());

JavaRDD<ExperimentForm> predf = CassandraJavaUtil.javaFunctions(dfexplistoriginal.javaRDD())
    .repartitionByCassandraReplica("mdb","experiment",experimentlist.size(),
        CassandraJavaUtil.someColumns("experimentid"),CassandraJavaUtil.mapToRow(ExperimentForm.class));
// 打印：RDD: PartNum 64
System.out.println("RDD: PartNum "+predf.getNumPartitions());

Dataset<ExperimentForm> newdfexplist =  sp.createDataset(predf.rdd(), ExpEncoder);
Dataset<Row> readydfexplist = newdfexplist.as(Encoders.STRING()).toDF("experimentid");
// 打印：DATASET: PartNum 64
System.out.println("DATASET: PartNum "+readydfexplist.rdd().getNumPartitions());

// 最终执行关联，但DAG中没有像往常一样标记为DirectJoin
Dataset<Row> metlistinitial = sp.read().format("org.apache.spark.sql.cassandra")
                .options(new HashMap<String, String>() {
                    {
                        put("keyspace", "mdb");
                        put("table", "experiment");
                    }
                })
                .load().select(col("experimentid"), col("description"), col("intensity"))
                .join(readydfexplist,"experimentid");

Spark UI观察到的现象 #

• 所有Stages概览：显示多个Stage，初始Stage任务数为4，后续Stage任务数增至64甚至更多
• Stage 0 DAG：展示从集合创建初始Dataset的流程，任务数4
• Stage 0 Metrics：对应初始Dataset的执行指标，分区数4
• Stage 1 DAG：包含repartitionByCassandraReplica相关的重分区操作
• Stage 1 Metrics：任务数变为64，且执行节点与预期的Cassandra存储节点不匹配

疑问 #

代码是否存在问题？为何会出现分区数异常和数据拉取节点错误的情况？

内容的提问来源于stack exchange，提问作者ktzan