Search code examples
apache-sparkcassandraspark-cassandra-connector

Spark-Cassandra: repartitionByCassandraReplica or converting dataset to JavaRDD and back do not maintain number of partitions?

So, I have a 16 node cluster where every node has Spark and Cassandra installed with a replication factor of 3 and spark.sql.shuffle.partitions of 96. I am using the Spark-Cassandra Connector 3.0.0 and I am trying to join a dataset with a cassandra table on the partition key, while also using .repartitionByCassandraReplica.

However repartitionByCassandraReplica is implemented only on RDDs so I am converting my dataset to JavaRDD, do the repartitionByCassandraReplica, then converting it back to dataset and do a Direct Join with the cassandra table. It seems though, that in the process of that the number of partitions is "changing" or is not as expected.

I am doing a PCA on 4 partition keys which have some thousands of rows and for which I know the nodes where they are stored according to nodetool getendpoints . It looks like not only the number of partitions is changing but also the nodes where data are pulled are not the ones that actually have the data. Below is the code.

//FYI experimentlist is a List<String> which is converted to Dataset,then JavaRDD, then partitioned
//according to repartitionByCassandraReplica and then back to Dataset. The table with which I want to
//join it, is called 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));
//Below prints 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));
//Below prints 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");
//Below prints DATASET: PartNum 64
System.out.println("DATASET: PartNum "+readydfexplist.rdd().getNumPartitions());

//and finally the DirectJoin which for some reason is not mentioned as DirectJoin in DAGs like other times
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");

Is the code wrong? Below are also some images from SparkUI the Stages Tab with DAGs. At first I have 4 tasks/partitions and after repartitionByCassandraReplica I get 64 or more. Why?

All the Stages:

enter image description here

Stage 0 DAG

enter image description here

Stage 0 Metrics

enter image description here

Stage 1 DAG

enter image description here

Stage 1 Some Metrics

enter image description here

Solution

  • Looks like the code I wrote above is not entirely correct! I managed to get repartitionByCassandraReplica working by just converting dataset to RDD, performing the repartitionByCassandraReplica doing the join with JoinWithCassandraTable and THEN converting back to dataset! Now it is indeed repartitioned on the nodes that actually have the data! Partitions are maintained between these conversions!