Spring Data Cassandra批量插入阈值适配：如何拆分批次？

当使用CassandraTemplate执行批量插入时，若单次批量实体过多，会触发Cassandra的BATCH_SIZE_FAIL_THRESHOLD_IN_KB（默认50KB）限制抛出异常。以下是几种可行的解决方案：

一、预先估算实体CQL大小，拆分批次 #

核心思路是先计算单个实体转换为CQL插入语句后的字节大小，再根据阈值反推每批可容纳的实体数量，将原列表拆分为多个子批次执行。

代码示例： #

import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
import com.datastax.oss.driver.api.querybuilder.insert.Insert;
import org.springframework.data.cassandra.core.CassandraTemplate;
import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntityMetadata;
import com.google.common.collect.Lists;
import java.nio.charset.StandardCharsets;
import java.util.List;

// 假设实体类为User
List<User> listOfEntities = ...;

// 取第一个实体（或大小最具代表性的实体）估算单条CQL大小
User sampleEntity = listOfEntities.get(0);
CqlIdentifier tableName = CassandraPersistentEntityMetadata.getTableName(User.class);
// 生成对应的INSERT语句
Insert insertQuery = QueryBuilder.insertInto(tableName)
        .values(cassandraTemplate.getConverter().write(sampleEntity, User.class));
String cql = insertQuery.getQueryString();
int singleEntityByteSize = cql.getBytes(StandardCharsets.UTF_8).length;

// Cassandra默认阈值为50KB，转换为字节；预留10%冗余空间避免超支
int thresholdInBytes = 50 * 1024;
int safeBatchSize = (int) (thresholdInBytes / singleEntityByteSize * 0.9);

// 拆分列表为多个批次
List<List<User>> batches = Lists.partition(listOfEntities, safeBatchSize);
for (List<User> batch : batches) {
    cassandraTemplate.batchOps().insert(batch).execute();
}

注意：如果实体字段内容长度差异较大（比如部分实体包含超长字符串），建议选取最大体积的实体来计算批次大小，避免个别批次超阈值。

二、异常捕获动态调整批次大小 #

如果实体大小差异极大，预先估算不够准确，可以采用"试错"方式：初始设置一个批次大小，执行时若触发批量过大异常，就减半批次大小重试，直到成功执行。

代码示例： #

import org.springframework.data.cassandra.core.CassandraTemplate;
import com.datastax.oss.driver.api.core.servererrors.QueryExecutionException;
import java.util.List;

List<User> listOfEntities = ...;
int currentBatchSize = 100; // 初始批次大小
int startIndex = 0;
int totalSize = listOfEntities.size();

while (startIndex < totalSize) {
    int endIndex = Math.min(startIndex + currentBatchSize, totalSize);
    List<User> currentBatch = listOfEntities.subList(startIndex, endIndex);
    
    try {
        cassandraTemplate.batchOps().insert(currentBatch).execute();
        startIndex = endIndex; // 成功则推进到下一批
    } catch (QueryExecutionException e) {
        // 判断是否是批量过大的异常
        if (e.getMessage() != null && e.getMessage().contains("Batch too large")) {
            currentBatchSize = currentBatchSize / 2;
            // 防止批次大小过小导致无限循环
            if (currentBatchSize < 1) {
                throw new RuntimeException("单个实体体积超过Cassandra批量阈值", e);
            }
        } else {
            // 其他异常直接抛出
            throw e;
        }
    }
}

三、修改Cassandra集群阈值（不推荐） #

可以修改Cassandra配置文件cassandra.yaml中的batch_size_fail_threshold_in_kb参数，调高阈值上限。但这种方式会影响整个集群的批量操作行为，过大的批量会增加节点压力，引发性能问题甚至节点崩溃，仅在明确业务场景允许且集群资源充足时考虑。

内容的提问来源于stack exchange，提问作者Soheil Rahsaz