Error uploading RDD to existing Cassandra table

Question

I extract content from a Cassandra table using the Dataframe connector functions. Afterwards I perform some filtering (no transformations) on the dataframe and then I want to write it again into an existing table. To do so I need to turn the DF into an RDD first. I do so using the following command:

To do this I used the following code:

results.rdd.map(row=> (row.get(0).asInstanceOf[String], row.get(1).asInstanceOf[String], row.get(2).asInstanceOf[String], row.get(3).asInstanceOf[java.util.UUID], row.get(4).asInstanceOf[String], row.get(5).asInstanceOf[String], row.get(6).asInstanceOf[Long], row.get(7).asInstanceOf[Set[String]], row.get(8).asInstanceOf[Array[Byte]], row.get(9).asInstanceOf[String], row.get(10).asInstanceOf[String], row.get(11).asInstanceOf[Set[String]], row.get(12).asInstanceOf[Set[String]], row.get(13).asInstanceOf[Set[String]], row.get(14).asInstanceOf[String], row.get(15).asInstanceOf[List[String]],row.get(16).asInstanceOf[String],row.get(17).asInstanceOf[String]))

I have checked and all the elements have their type set appropriately. Now I want to upload this RDD to Cassandra using results.saveToCassandra("labels", "results", SomeColumns(...)) but I get the following error.

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 73.0 failed 4 times, most recent failure: Lost task 0.3 in stage 73.0 (TID 70606, hdp-worker-5.cloud.mwn.de, executor 172): java.lang.ClassCastException: scala.collection.mutable.WrappedArray$ofRef cannot be cast to scala.collection.immutable.Set
    at $$$$b42ed3d02f91ffa45dcf288dd693450$$$$$anonfun$1.apply(<console>:65)
    at $$$$b42ed3d02f91ffa45dcf288dd693450$$$$$anonfun$1.apply(<console>:65)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
    at com.datastax.spark.connector.util.CountingIterator.next(CountingIterator.scala:16)
    at com.datastax.spark.connector.writer.GroupingBatchBuilder.next(GroupingBatchBuilder.scala:106)
    at com.datastax.spark.connector.writer.GroupingBatchBuilder.next(GroupingBatchBuilder.scala:31)
    at scala.collection.Iterator$class.foreach(Iterator.scala:893)

The thing is that the Cassandra type for several columns is Set which means my scala type needs to be a type of Set[String] (or TreeSet or HashSet - I tested them all) and I get this error. How do I format my data types correctly to upload to Cassandra?

Edit:

I updated the code according to the first suggestion:

results.rdd.map(row=> (row.getString(0), row.getString(1), row.getString(2), row.get[java.util.UUID](3), row.getString(4), row.getString(5), row.get[java.util.Date](6), row.get[Seq[String]](7).toSet, row.get[Array[Byte]](8), row.getString(9), row.getString(10), row.getSeq[String](11).toSet, row.get[Seq[String]](12).toSet, row.get[Seq[String]](13).toSet, row.getString(14), row.get[List[String]](15),row.getString(16),row.getString(17)))

But I keep getting errors related to several of the entries:

<console>:65: error: method get: (i: Int)Any does not take type parameters.

Answer 1

asInstance won't turn your data into something it is not. Instead you should convert it (and in general use typed getters not asInstanceOf):

results.rdd.map(row => (
    row.getString(0),
    ...
    row.getSeq[String](7).toSet,
    ...,
    row.getSeq[String](11).toSet,
    ...
))

There might be other errors in this code, especially:

row.get(3).asInstanceOf[java.util.UUID]

looks funky, as Spark has no UUID type.

Related Spark extracting values from a Row