Using the Spark context

Python support for loading cassandraTables 

Python supports loading cassandraTables from a Spark Streaming context and saving a DStream to Cassandra.

Reading column values 

You can read columns in a Cassandra table using the get methods of the CassandraRow object. The get methods access individual column values by column name or column index. Type conversions are applied on the fly. Use getOption variants when you expect to receive Cassandra null values.

Reading collections 

You can read collection columns in a Cassandra table using the get methods of the CassandraRow object. The get methods access the collection column and returns a corresponding Scala collection.

Assuming you set up the test keyspace earlier, follow these steps to access a Cassandra collection.

1. In the test keyspace, set up a collection set using cqlsh.

   CREATE TABLE test.users (
     username text PRIMARY KEY, emails SETtext);
   
   INSERT INTO test.users (username, emails) 
     VALUES ('someone', {'someone@email.com', 's@email.com'});

2. If Spark is not running, start the Spark shell. Do not use sudo to start the shell.

   $ bin/dse spark

   The Welcome to Spark output and scala prompt appears.

3. Define a CassandraRDD[CassandraRow] to access the collection set.

   val row = sc.cassandraTable("test", "users").toArray.apply(0)
   

   row: com.datastax.spark.connector.CassandraRow = CassandraRow{username: someone, 
   emails: {s@email.com,someone@email.com}}

4. Query the collection set in Cassandra from Spark.

   row.getList[String]("emails")
   

   res2: Vector[String] = Vector(s@email.com, someone@email.com)
   

   row.get[List[String]]("emails")
   

   res3: List[String] = List(s@email.com, someone@email.com)
   

   row.get[Seq[String]]("emails")
   

   res4: Seq[String] = List(s@email.com, someone@email.com)
   

   row.get[IndexedSeq[String]]("emails")
   

   res5: IndexedSeq[String] = Vector(s@email.com, someone@email.com)
   

   row.get[Set[String]]("emails")
   

   res6: Set[String] = Set(s@email.com, someone@email.com)
   

   row.get[String]("emails")
   

   res7: String = {s@email.com,someone@email.com}

Restricting the number of fetched columns 

For performance reasons, you should not fetch columns you don't need. You can achieve this with the select method:

To restrict the number of fetched columns:

val row = sc.cassandraTable("test", "users").select("username").toArray

row: Array[com.datastax.spark.connector.CassandraRow] = Array(CassandraRow{username: someone})

Mapping rows to tuples and case classes 

Instead of mapping your Cassandra rows to objects of the CassandraRow class, you can directly unwrap column values into tuples of the desired type.

To map rows to tuples:

sc.cassandraTable[(String, Int)]("test", "words").select("word", "count").toArray

res9: Array[(String, Int)] = Array((bar,20), (foo,10))

sc.cassandraTable[(Int, String)]("test", "words").select("count", "word").toArray

res10: Array[(Int, String)] = Array((20,bar), (10,foo))

Define a case class with properties of the same name as the Cassandra columns. For multi-word column identifiers, separate each word using an underscore in Cassandra, and use camel case abbreviation on the Scala side.

To map rows to case classes:

case class WordCount(word: String, count: Int)

defined class WordCount

scala> sc.cassandraTable[WordCount]("test", "words").toArray

res14: Array[WordCount] = Array(WordCount(bar,20), WordCount(foo,20))

You can name columns in Cassandra using these conventions:

• Use the underscore convention and lowercase letters. (Recommended)
• Use the camel case convention, exactly the same as properties in Scala.

The following examples show valid column names.

Mapping rows to objects with a user-defined function 

Invoke as on the CassandraRDD to map every row to an object of a different type. Contrary to map, as expects a function having the same number of arguments as the number of columns to be fetched. Invoking as in this way performs type conversions. Using as to directly create objects of a particular type eliminates the need to create CassandraRow objects and also decreases garbage collection pressure.

To map columns using a user-defined function:

val table = sc.cassandraTable("test", "words")

table: com.datastax.spark.connector.rdd.CassandraRDD[com.datastax.spark.connector.
CassandraRow] = CassandraRDD[9] at RDD at CassandraRDD.scala:47

val total = table.select("count").as((c: Int) => c).sum

total: Double = 30.0

val frequencies = table.select("word", "count").as((w: String, c: Int) => (w, c / total)).toArray

frequencies: Array[(String, Double)] = Array((bar,0.6666666666666666), (foo,0.3333333333333333))

Filtering rows on the server 

To filter rows, you can use the filter transformation provided by Spark. Filter transformation fetches all rows from Cassandra first and then filters them in Spark. Some CPU cycles are wasted serializing and deserializing objects excluded from the result. To avoid this overhead, CassandraRDD has a method that passes an arbitrary CQL condition to filter the row set on the server.