Query cassandra from spark executor

Question

I have a streaming app off of kafka, and I was wondering if there was a way to do a range query from inside a map function?

I group the messages from kafka by time range and key, and then based on those time ranges and keys I want to pull data from cassandra into that dstream.

Something like:

lookups
  .map(lookup => ((lookup.key, lookup.startTime, lookup.endTime), lookup))
  .groupByKey()
  .transform(rdd => {
    val cassandraSQLContext = new CassandraSQLContext(rdd.context)
    rdd.map(lookupPair => {
      val tableName = //variable based on lookup
      val startTime = aggLookupPair._1._2
      val endTime = aggLookupPair._1._3

      cassandraSQLContext
        .cassandraSql(s"SELECT * FROM ${CASSANDRA_KEYSPACE}.${tableName} WHERE key=${...} AND start_time >= ${startTime} AND start_time < ${endTime};")
        .map(row => {
           //match to {
            case /*case 1*/ => new object1(row)
            case /*case 2*/ =>new object2(row)
          }
        })
        .collect()
    })
  })

This gives me a null pointer exception:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 59.0 failed 1 times, most recent failure: Lost task 0.0 in stage 59.0 (TID 63, localhost): java.lang.NullPointerException
at org.apache.spark.sql.SQLContext.parseSql(SQLContext.scala:231)
at org.apache.spark.sql.cassandra.CassandraSQLContext.cassandraSql(CassandraSQLContext.scala:70)
at RollupFineGrainIngestionService$$anonfun$11$$anonfun$apply$2.apply(MyFile.scala:130)
at RollupFineGrainIngestionService$$anonfun$11$$anonfun$apply$2.apply(MyFile.scala:123)
at scala.collection.Iterator$$anon$11.next(Iterator.scala:370)
at org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:285)
at org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:171)
at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:78)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:268)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)

I've also tried to ssc.cassandraTable(CASSANDRA_KEYSPACE, tableName).where("key = ?", ...)... but spark crashes when trying to access StreamingContext inside of a map.

If anyone has any suggestions, I would appreciate it. Thanks!

RussS RussS · Accepted Answer · 2016-07-14T21:04:11

You may want to use joinWithCassandraTable if your query is based off of a partition key.

But if you need more flexibility

CassandraConnector(sc.getConf).withSessionDo( session => ...)

Will allow you to access the session pool on the executor and execute whatever you want without managing connections. The code is all serializable and can be placed within maps.

Query cassandra from spark executor

1 Answers