CODEWITHSUNDEEP
cassandraapache-sparkapache-spark-sql
Nicky in the Clouds
Nicky in the Clouds

Reputation: 210

Spark SQL Cassandra how to handle timestamp empty value?

I'm currently using an Apache Cassandra 2.1.2 cluster with Spark 1.2.0 connector. For some initial tests, I need to select some rows from a Cassandra table through Spark SQL commands within the spark-shell.

We use a table called tabletest in the keyspace ks. This table contains for example an id (bigint) and a ts (timestamp).

here is my spark script :

import com.datastax.spark.connector._
import org.apache.spark.sql.cassandra.CassandraSQLContext
val cc = new CassandraSQLContext(sc)
cc.setKeyspace("ks")
val rdd = cc.sql("SELECT id,ts FROM tabletest LIMIT 100")
rdd.toArray.foreach(println)

when I execute this script through the command :

spark-shell -i myscript

everything is ok till a row contains an empty value for the ts cell. If there is a row with an empty value for ts, I got several exceptions related to the fact spark is waiting a long value (8 bytes) and get no byte. I got the same issue even if I try to count the number of rows without displaying the rows.

15/01/29 15:21:35 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)
com.datastax.driver.core.exceptions.InvalidTypeException: Invalid 64-bits long value, expecting 8 bytes but got 0
  at com.datastax.driver.core.TypeCodec$LongCodec.deserializeNoBoxing(TypeCodec.java:452)
  at com.datastax.driver.core.TypeCodec$DateCodec.deserialize(TypeCodec.java:826)
  at com.datastax.driver.core.TypeCodec$DateCodec.deserialize(TypeCodec.java:748)
  at com.datastax.driver.core.DataType.deserialize(DataType.java:606)
  at com.datastax.spark.connector.AbstractGettableData$.get(AbstractGettableData.scala:88)
  at org.apache.spark.sql.cassandra.CassandraSQLRow$$anonfun$fromJavaDriverRow$1.apply$mcVI$sp(CassandraSQLRow.scala:42)
  at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:141)
  at org.apache.spark.sql.cassandra.CassandraSQLRow$.fromJavaDriverRow(CassandraSQLRow.scala:41)
  at org.apache.spark.sql.cassandra.CassandraSQLRow$CassandraSQLRowReader$.read(CassandraSQLRow.scala:49)
  at org.apache.spark.sql.cassandra.CassandraSQLRow$CassandraSQLRowReader$.read(CassandraSQLRow.scala:46)
  at com.datastax.spark.connector.rdd.CassandraRDD$$anonfun$13.apply(CassandraRDD.scala:378)
  at com.datastax.spark.connector.rdd.CassandraRDD$$anonfun$13.apply(CassandraRDD.scala:378)
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
  at scala.collection.Iterator$$anon$13.next(Iterator.scala:372)
  at com.datastax.spark.connector.util.CountingIterator.next(CountingIterator.scala:13)
  at scala.collection.Iterator$$anon$10.next(Iterator.scala:312)
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
  at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
  at org.apache.spark.util.collection.ExternalSorter.spillToPartitionFiles(ExternalSorter.scala:366)
  at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:211)
  at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:65)
  at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)
  at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
  at org.apache.spark.scheduler.Task.run(Task.scala:56)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
  at java.lang.Thread.run(Thread.java:745)

How could I handle such empty value, do I have to use some functions within my SQL query to replace empty value with default value or could I use some methods or parameters in my script to allow spark to handle such empty values ?

Thanks for your help,

Best

Nicolas

Upvotes: 0

Views: 1283

Answers (1)

Stefan Podkowinski
Stefan Podkowinski

Reputation: 5249

Null values can be handled by asking for options. I'd suggest you give it a try by replacing the println function in your script with a custom function that will use the row values directly. Something like ((row) => println(row.getLong("id") + "," + row.getLongOption("ts"))

Upvotes: -1

Related Questions