Volume insert in Cassandra UDT using spark

Question

Volume insert in Cassandra UDT using spark

I use spark to mass insert Facebook data for analysis. I have a comment as UDT in cassandra. the fbpost table, which has a set of comments in the form of a column. Below is a diagram.

CREATE TYPE analytics.comment (
commentid text,
commenttext varchar,
username text,
commentdatetime timestamp );

CREATE TABLE analytics.posts (
postid text,
username text,
comments set<frozen<comment>>,
datetime timestamp,
posttext varchar,
PRIMARY KEY (datetime, username)

From a spark job written in Java, I create a JavaRDD (such as FBposts) that I need to save in a Cassandra table. I have pojos for fbpost and comments.

 javaFunctions(fbPostFromMysql).writerBuilder("analytics", "posts", fbPostWriter).saveToCassandra();

Maybe I am not doing the mapping for my Java pojo comment to UDTValue correctly , so I get a TypeConversionException below, this is a stacktrace

localhoatastax.spark.connector.types.TypeConversionException: Cannot convert object Comment [commentId=642528915901340_642701759217389, commenttext=Prix, username=Angel Nannousa, commentdatetime=2016-07-0 
class com.prophecy.spark.cassandra.model.Comment to com.datastax.driver.core.UDTValue

Here is the java code for rowwriter

public class FbPostRowWriter implements RowWriter<FbPost> {
private static final long serialVersionUID = 1L;
private static RowWriter<FbPost> writer = new FbPostRowWriter();

// Factory
public static class FbPostRowWriterFactory implements RowWriterFactory<FbPost>, Serializable{
    private static final long serialVersionUID = 1L;

    @Override
    public RowWriter<FbPost> rowWriter(TableDef tableDef, IndexedSeq<ColumnRef> arg1) {
        return writer;
    }
}

@Override
public Seq<String> columnNames() {      
    return scala.collection.JavaConversions.asScalaBuffer(FbPost.columns()).toList();
}

@Override
public void readColumnValues(FbPost summary, Object[] buffer) {
    buffer[0] = summary.getPostId();
    buffer[1] = summary.getUsername();
    buffer[3] = summary.getDatetime();
    buffer[4] = summary.getPosttext();  
    buffer[2] = summary.getComments();
    }
}