CasBuilder not serializable

[lvicentesanchez]

I'm trying to use calliope with spark streaming and I'm getting a 'Task not serializable' exception. I'm trying to use cql3SaveToCassandra this way,

    reducedByKey.foreachRDD { 
      _.leftOuterJoin(casRdd).map {
        case (key, (value, optValue)) ⇒
          (key, optValue.map(_ + value).getOrElse(value))
      }.cql3SaveToCassandra(cas)
    }

If I remove 'cql3SaveToCassandra(cas)', there is no exception so the problem is that CasBuilder.cql3 (and probably the whole CasBuilder object) is not serializable.

I only have this problem while trying to use 0.9.4-EA-SNAPSHOT; with spark 0.9 and 0.9.0-C2-EA, I didn't have that problem... so I don't know if there was a change on spark or on the SNAPSHOT that is causing that problem.

[lvicentesanchez]

I have checked the source code and I'm really confused... as far as I can see Cql3CasBuilder is serializable so... it shouldn't be problem :S

[lvicentesanchez]

I think I might have found the problem...

def cql3SaveToCassandra(cas: Cql3CasBuilder)
                         (implicit keyMarshaller: U => CQLRowKeyMap, rowMarshaller: U => CQLRowValues)

My key and row marshallers look like this:

  implicit def RevenueBucketKeyMarshaller(row: ((BigInt, String), BigDecimal)): CQLRowKeyMap = row match {
    case ((key1, key2), _) ⇒
      Map("time" -> DataType.bigint().serialize(key1.bigInteger), "currency" -> key2)
  }

  implicit def RevenueBucketRowMarshaller(row: ((BigInt, String), BigDecimal)): CQLRowValues = row match {
    case (_, value) ⇒
      List(DataType.decimal().serialize(value.bigDecimal))
  }

if DataType.decimal().serialize and DataType.bigint().serialize are not serializable, that would explain the problems I'm having.

[lvicentesanchez]

Problem solve... it's not a calliope issue; I was mixing my implicit conversions with my Job class that was not serializable. The error message wasn't too helpful :(