目录
spring data 操作 Cassandra
添加依赖和配置
- 添加依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-cassandra</artifactId>
</dependency>
- 添加配置
spring.data.cassandra.keyspace-name=myKeySpace
spring.data.cassandra.contact-points=myContactPoints
spring.data.cassandra.port=9042
spring.data.cassandra.schema-action=NONE
定义key, data 和 建表
此处以counter表为例, 表示用户在某一个类别下的点击量。
@PrimaryKeyClass
public class UserCategoryKey implements Serializable {
@PrimaryKeyColumn(name = "user_id", ordinal = 0, type = PrimaryKeyType.PARTITIONED)
private String userId;
@PrimaryKeyColumn(name = "category_id", ordinal = 1, type = PrimaryKeyType.CLUSTERED)
private Integer categoryId;
}
@Table("user_category")
public class UserCategory {
@PrimaryKey
private UserCategoryKey key;
@Column("click_cnt")
private Long clickCnt;
}
创建表的schema：
-- Cannot set default_time_to_live on a table with counters
CREATE TABLE user_category
(
user_id text,
category_id int,
click_cnt counter
PRIMARY KEY ((user_id), category_id)
);
限制:
- 非counter字段必须包含在主键中。
使用repository操作数据
@Repository
public interface UserCategoryRepository extends CassandraRepository<UserCategory, UserCategoryKey> {
@Modifying
@Transactional
@Query(value = "update user_category set click_cnt = click_cnt + :cnt where " +
"user_id = :#{#key.userId} and category_id = :#{#key.categoryId}")
public void incClickCnt(@Param("key") UserCategoryKey key, @Param("cnt") Long cnt);
}
使用CassandraOperations 操作数据
CassandraOperations 可以直接注入：
@Autowired
protected CassandraOperations cassandraTemplate;
/**
* update user_category set click_cnt = click_cnt + 1 where user_id='user1' and category_id = 2001
*/
public Boolean incClickCounter(UserCategoryKey key, Long cnt) {
try {
//update table
Update update = QueryBuilder.update("user_category");
//set
update.with(QueryBuilder.incr("click_cnt", cnt));
//where conditions
update.where(QueryBuilder.eq("user_id", key.getUserId()))
.and(QueryBuilder.eq("category_id", key.getCategoryId()));
cassandraTemplate.getCqlOperations().execute(update);
return true;
} catch (Exception e) {
LOG.error("incClickCounter exception, msg:{}", e.getMessage(), e);
}
return false;
}
踩坑二三事
1. 只能对counter字段进行加减操作
如果对非counter字段进行加减操作，报错如下：
Query; CQL [update user_category set click_cnt = click_cnt + ? where user_id = ? and category_id = ?]; Invalid operation (click_cnt = click_cnt + ?) for non counter column click_cnt; nested exception is com.datastax.driver.core.exceptions.InvalidQueryException: Invalid operation (click_cnt = click_cnt + ?) for non counter column click_cnt
2. 参数位置要求
对于 click_cnt = click_cnt + :cnt 如果参数放在前面，即click_cnt = :cnt + click_cnt. 如下语句 update user_category set click_cnt = :cnt + click_cnt where user_id = 'user1' and category_id = 1, 则报错：
Query; CQL [update user_category set click_cnt = ? + click_cnt where user_id = ? and category_id = ?]; Invalid operation (click_cnt = ? - click_cnt) for non list column click_cnt; nested exception is com.datastax.driver.core.exceptions.InvalidQueryException: Invalid operation (click_cnt = ? - click_cnt) for non list column click_cnt
4. counter操作的参数需为long型
如下，将方法中参数cnt定义为Integer：
@Query(value = "update user_category set click_cnt = click_cnt + :cnt where user_id = 'user1' and category_id = 1")
public void incClickCnt(@Param("cnt") Integer cnt);
报错如下：
Query; CQL [update user_category set click_cnt = click_cnt + ? where user_id=? and category_id = ?]; Expected 8 or 0 byte long (4); nested exception is com.datastax.driver.core.exceptions.InvalidQueryException: Expected 8 or 0 byte long (4)
5. in语句不要加小括号
如下：
@Query(value = "update user_category set click_cnt = click_cnt + 1 where user_id='user1' and category_id in :categoryIdList")
public void bathUpdate(@Param("categoryIdList") List<Integer> categoryIdList);
若改为news_id in (:newsIdList), 则会报错：
org.springframework.data.cassandra.CassandraInvalidQueryException: Query; CQL [update user_category set click_cnt=click_cnt+1 where user_id=? and category_id in (?)]; Expected 4 or 0 byte int (28); nested exception is com.datastax.driver.core.exceptions.InvalidQueryException: Expected 4 or 0 byte int (28 ）
6. 关于CassandraOperations
如果在unit test 中直接注入CassandraOperations测试，会报JmxReporter相关的错。但是如果在业务代码中直接注入，不会有问题。如果需要在单测中使用CassandraOperations，可以手动build一个Cluster，并禁用 JmxReporter 和 metrics。
参考：
Cassandra 计数器counter类型和它的限制
Cassandra数据建模
最后：
没有很多时间逐个搞清楚每个问题的深层次原因啦，记录下踩的坑，以鉴他人~