ClassCastException: java.util.Date cannot be cast to java.sql.Timestamp

[oded peer]

The HTML layout is much nicer where I originally posted this bug - https://github.com/datastax/spark-cassandra-connector/issues/512



I created a table in Cassandra and inserted data:

create table timestamp_conversion_bug (k int, v int, d timestamp, primary key(k,v));
insert into timestamp_conversion_bug (k, v, d) values (1, 1, '2015-01-03 15:13');
insert into timestamp_conversion_bug (k, v, d) values (1, 2, '2015-01-03 16:13');
insert into timestamp_conversion_bug (k, v, d) values (1, 3, '2015-01-03 17:13');
insert into timestamp_conversion_bug (k, v, d) values (1, 4, '2015-01-03 18:13');



I then queried the table using the connector and Spark SQL:

package poc;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SchemaRDD;
import org.apache.spark.sql.cassandra.CassandraSQLContext;
import org.apache.spark.sql.catalyst.expressions.Row;

import com.datastax.bdp.spark.DseSparkConfHelper;

public class TimestampConversionBug {

public static void main( String[] args ) {
SparkConf conf = DseSparkConfHelper.enrichSparkConf(new SparkConf().setAppName( "TimestampConversionBug" ));
JavaSparkContext jsc = new JavaSparkContext(conf);
CassandraSQLContext cassandraContext = new CassandraSQLContext(jsc.sc());
String keyspace = "test";
cassandraContext.setKeyspace( keyspace );

SchemaRDD rdd = cassandraContext.sql( "select k, min(d), max(d) from test.timestamp_conversion_bug group by k" );
JavaRDD<Row> jrdd = rdd.toJavaRDD();
for ( Row row : jrdd.collect() ) {
System.out.println( row );
}
}
}



This results in the following stacktrace:

15/01/15 10:19:39 ERROR TaskSetManager: Task 0 in stage 1.0 failed 4 times; aborting job
Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 1.0 failed 4 times, most recent failure: Lost task 0.3 in stage 1.0 (TID 3, 127.0.0.1): java.lang.ClassCastException: java.util.Date cannot be cast to java.sql.Timestamp
org.apache.spark.sql.catalyst.types.TimestampType$$anon$1.compare(dataTypes.scala:166)
scala.math.Ordering$class.gt(Ordering.scala:97)
org.apache.spark.sql.catalyst.types.TimestampType$$anon$1.gt(dataTypes.scala:166)
org.apache.spark.sql.catalyst.expressions.GreaterThan$$anonfun$eval$4.apply(predicates.scala:187)
org.apache.spark.sql.catalyst.expressions.GreaterThan$$anonfun$eval$4.apply(predicates.scala:187)
org.apache.spark.sql.catalyst.expressions.Expression.c2(Expression.scala:217)
org.apache.spark.sql.catalyst.expressions.GreaterThan.eval(predicates.scala:187)
org.apache.spark.sql.catalyst.expressions.MinFunction.update(aggregates.scala:114)
org.apache.spark.sql.execution.Aggregate$$anonfun$execute$1$$anonfun$7.apply(Aggregate.scala:167)
org.apache.spark.sql.execution.Aggregate$$anonfun$execute$1$$anonfun$7.apply(Aggregate.scala:151)
org.apache.spark.rdd.RDD$$anonfun$13.apply(RDD.scala:596)
org.apache.spark.rdd.RDD$$anonfun$13.apply(RDD.scala:596)
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:262)
org.apache.spark.rdd.RDD.iterator(RDD.scala:229)
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:262)
org.apache.spark.rdd.RDD.iterator(RDD.scala:229)
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)
org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
org.apache.spark.scheduler.Task.run(Task.scala:54)
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
java.lang.Thread.run(Thread.java:745)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1185)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1174)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1173)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1173)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:688)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:688)
at scala.Option.foreach(Option.scala:236)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:688)
at org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1391)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498)
at akka.actor.ActorCell.invoke(ActorCell.scala:456)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237)
at akka.dispatch.Mailbox.run(Mailbox.scala:219)
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)



Note that when I execute the query in the interactive DSE Spark command line the query works:

scala> sc.sql("select k, min(d), max(d) from test.timestamp_conversion_bug group by k");
OK
15/01/15 10:20:45 WARN LoadSnappy: Snappy native library not loaded
res4: Seq[String] = Buffer(1 2015-01-03 15:13:00 2015-01-03 18:13:00)

[Phil Kallos]

I am also seeing something similar with spark 1.1.1 and datastax cassandra-spark 1.1.1

scala> cc.sql("select field1,ts from cf.table order by ts desc").take(100).foreach(println)

Yields me a

: java.lang.ClassCastException: java.util.Da

te cannot be cast to java.sql.Timestamp
org.apache.spark.sql.catalyst.types.TimestampType$$anon$1.compare(dataTypes.scala:166)

scala.math.Ordering$$anon$4.compare(Ordering.scala:111)
org.apache.spark.sql.catalyst.expressions.RowOrdering.compare(Row.scala:260)
org.apache.spark.sql.catalyst.expressions.RowOrdering.compare(Row.scala:238)
org.apache.spark.util.collection.Utils$$anon$1.compare(Utils.scala:35)
com.google.common.collect.Ordering.max(Ordering.java:572)
com.google.common.collect.Ordering.leastOf(Ordering.java:688)
org.apache.spark.util.collection.Utils$.takeOrdered(Utils.scala:37)
org.apache.spark.rdd.RDD$$anonfun$takeOrdered$1.apply(RDD.scala:1136)
org.apache.spark.rdd.RDD$$anonfun$takeOrdered$1.apply(RDD.scala:1133)

org.apache.spark.rdd.RDD$$anonfun$13.apply(RDD.scala:596)
org.apache.spark.rdd.RDD$$anonfun$13.apply(RDD.scala:596)
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:262)
org.apache.spark.rdd.RDD.iterator(RDD.scala:229)

org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62)
org.apache.spark.scheduler.Task.run(Task.scala:54)
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:178)

java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
java.lang.Thread.run(Thread.java:745)

[Piotr Kołaczkowski]

Hey Alex, can you take a look it?
Thanks,
Piotr

To unsubscribe from this group and stop receiving emails from it, send an email to spark-connector-...@lists.datastax.com.

[julia.w...@gmail.com]

Hi All,

I also get the same error when using a timestamp in the Cassandra Table. My table and data looks like below -

CREATE TABLE test.issuer_dispute_entrydate (
country text,
bid int,
bin int,
jursdctn text,
fund_src text,
crd_prsd_typ text,
tot_dspt_sbmtd_cnt_entry_dt timestamp,
PRIMARY KEY ((country ,bin), bid ,jursdctn ,fund_src ,crd_prsd_typ)
);

insert into test.issuer_dispute_entrydate (country ,bin ,bid ,jursdctn ,fund_src ,crd_prsd_typ,tot_dspt_sbmtd_cnt_entry_dt)
values ('US',1,1,'a','b','c',toTimestamp(now()));
insert into test.issuer_dispute_entrydate (country ,bin ,bid ,jursdctn ,fund_src ,crd_prsd_typ,tot_dspt_sbmtd_cnt_entry_dt)
values ('CA',2,1,'a','b','c',null);

My Java class for this is -

import java.sql.Timestamp;
public class IssuerDisputeTotalAggEntryDate implements Serializable{
private String country;
private Integer bin;
private Integer bid;
private String jursdctn;
private String fund_src;
private String crd_prsd_typ;
private Timestamp tot_dspt_sbmtd_cnt_entry_dt;
public String getCountry() {
return country;
}
public void setCountry(String country) {
this.country = country;
}
public Integer getBin() {
return bin;
}
public void setBin(Integer bin) {
this.bin = bin;
}
public Integer getBid() {
return bid;
}
public void setBid(Integer bid) {
this.bid = bid;
}
public String getJursdctn() {
return jursdctn;
}
public void setJursdctn(String jursdctn) {
this.jursdctn = jursdctn;
}
public String getFund_src() {
return fund_src;
}
public void setFund_src(String fund_src) {
this.fund_src = fund_src;
}
public String getCrd_prsd_typ() {
return crd_prsd_typ;
}
public void setCrd_prsd_typ(String crd_prsd_typ) {
this.crd_prsd_typ = crd_prsd_typ;
}
public Timestamp getTot_dspt_sbmtd_cnt_entry_dt() {
return tot_dspt_sbmtd_cnt_entry_dt;
}
public void setTot_dspt_sbmtd_cnt_entry_dt( Timestamp tot_dspt_sbmtd_cnt_entry_dt) {
this.tot_dspt_sbmtd_cnt_entry_dt = tot_dspt_sbmtd_cnt_entry_dt ;
}
}

Exception -

17/10/05 01:09:26 INFO CodeGenerator: Code generated in 14.929731 ms
Exception in thread "main" java.lang.IllegalArgumentException: Unsupported type: java.sql.Timestamp
at com.datastax.spark.connector.types.TypeConverter$.forCollectionType(TypeConverter.scala:997)
at com.datastax.spark.connector.types.TypeConverter$.forType(TypeConverter.scala:1010)
at com.datastax.spark.connector.types.TypeConverter$.forType(TypeConverter.scala:1029)
at com.datastax.spark.connector.rdd.reader.GettableDataToMappedTypeConverter.converter(GettableDataToMappedTypeConverter.scala:108)
at com.datastax.spark.connector.rdd.reader.GettableDataToMappedTypeConverter.com$datastax$spark$connector$rdd$reader$GettableDataToMappedTypeConverter$$converter(GettableDataToMappedTypeConverter.scala:117)
at com.datastax.spark.connector.rdd.reader.GettableDataToMappedTypeConverter$$anonfun$7.apply(GettableDataToMappedTypeConverter.scala:184)
at com.datastax.spark.connector.rdd.reader.GettableDataToMappedTypeConverter$$anonfun$7.apply(GettableDataToMappedTypeConverter.scala:181)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)

What is the recommended data type to map Cassandra Time Stamps in Spark

Thanks,
Manish