Я пытаюсь вставить простую строку в таблицу в Cassandra 2.0.5, Storm версии 0.9.0.1.
Мой тест выглядит следующим образом:
У меня есть таблица, состоящая из столбца id (int) и предложения (text). id - это первичный ключ.
Мой носик генерирует предложения, и я добавляю идентификатор (статическое приращение кода).
Это моя топология:
TridentTopology topology = new TridentTopology();
StateFactory cassandraStateFactory = CassandraMapState.nonTransactional(options);
Fields fields = new Fields("id", "sentence");
MyTridentTupleMapper tupleMapper = new MyTridentTupleMapper(keyspace, fields);
CassandraUpdater updater = new CassandraUpdater(tupleMapper);
TridentState wordCounts = topology.newStream("spout1", spout)
.each(new Fields("sentence"), new AddId(), new Fields("id"))
.partitionPersist(cassandraStateFactory, fields, updater);
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("test", config, topology.build());
Код для MyTridentTupleMapper:
У меня следующее исключение:
2014-02-08 22:20:14 ERROR executor:0 -
java.lang.RuntimeException: java.lang.ClassCastException: storm.trident.state.map.SnapshottableMap cannot be cast to com.hmsonline.storm.cassandra.trident.CassandraState
at backtype.storm.utils.DisruptorQueue.consumeBatchToCursor(DisruptorQueue.java:90)
at backtype.storm.utils.DisruptorQueue.consumeBatchWhenAvailable(DisruptorQueue.java:61)
at backtype.storm.disruptor$consume_batch_when_available.invoke(disruptor.clj:62)
at backtype.storm.daemon.executor$fn__3498$fn__3510$fn__3557.invoke(executor.clj:730)
at backtype.storm.util$async_loop$fn__444.invoke(util.clj:403)
at clojure.lang.AFn.run(AFn.java:24)
at java.lang.Thread.run(Thread.java:744)
Я не уверен, почему он это возвращает, и был бы признателен за помощь.