cassandra-spring ingest command doesn't work

Question

I've set up a cassandra cluster and work with the spring-cassandra framework 1.53. (http://docs.spring.io/spring-data/cassandra/docs/1.5.3.RELEASE/reference/html/)

I want to write millions of datasets into my cassandra cluster. The solution with executeAsync works good but the "ingest" command from the spring framework sounds interesting aswell.

The ingest method takes advantage of static PreparedStatements that are only prepared once for performance. Each record in your data set is bound to the same PreparedStatement, then executed asynchronously for high performance.

My code:

List<List<?>> session_time_ingest = new ArrayList<List<?>>();
for (Long tokenid: listTokenID) {
List<Session_Time_Table> tempListSessionTimeTable = repo_session_time.listFetchAggregationResultMinMaxTime(tokenid);
session_time_ingest.add(tempListSessionTimeTable);
}

cassandraTemplate.ingest("INSERT into session_time (sessionid, username, eserviceid, contextroot," +
                " application_type, min_processingtime, max_processingtime, min_requesttime, max_requesttime)" +
                " VALUES(?,?,?,?,?,?,?,?,?)", session_time_ingest);

Throws exception:

`Exception in thread "main" com.datastax.driver.core.exceptions.CodecNotFoundException: Codec not found for requested operation: [varchar <-> ...tracking.Tables.Session_Time_Table]
at com.datastax.driver.core.CodecRegistry.notFound(CodecRegistry.java:679)
at com.datastax.driver.core.CodecRegistry.createCodec(CodecRegistry.java:540)
at com.datastax.driver.core.CodecRegistry.findCodec(CodecRegistry.java:520)
at com.datastax.driver.core.CodecRegistry.codecFor(CodecRegistry.java:470)
at com.datastax.driver.core.AbstractGettableByIndexData.codecFor(AbstractGettableByIndexData.java:77)
at com.datastax.driver.core.BoundStatement.bind(BoundStatement.java:201)
at com.datastax.driver.core.DefaultPreparedStatement.bind(DefaultPreparedStatement.java:126)
at org.springframework.cassandra.core.CqlTemplate.ingest(CqlTemplate.java:1057)
at org.springframework.cassandra.core.CqlTemplate.ingest(CqlTemplate.java:1077)
at org.springframework.cassandra.core.CqlTemplate.ingest(CqlTemplate.java:1068)
at ...tracking.SessionAggregationApplication.main(SessionAggregationApplication.java:68)`

I coded exactly like in the spring-cassandra doku.. I've no idea how to map the values of my object to the values cassandra expects?!

Answer 1

Your Session_Time_Table class is probably a mapped POJO, but ingest methods do not use POJO mapping.

Instead you need to provide a matrix where each row contains as many arguments as there are variables to bind in your prepared statement, something along the lines of:

List<List<?>> rows = new ArrayList<List<?>>();

for (Long tokenid: listTokenID) {
    Session_Time_Table obj = ... // obtain a Session_Time_Table instance
    List<Object> row = new ArrayList<Object>();
    row.add(obj.sessionid);
    row.add(obj.username);
    row.add(obj.eserviceid);
    // etc. for all bound variables
    rows.add(row);
}

cassandraTemplate.ingest(
    "INSERT into session_time (sessionid, username, eserviceid, " +
    "contextroot, application_type, min_processingtime, " +
    "max_processingtime, min_requesttime, max_requesttime) " +
    "VALUES(?,?,?,?,?,?,?,?,?)", rows);